1use std::{
19 collections::{HashMap, HashSet},
20 fmt::Debug,
21 num::NonZeroU32,
22 sync::{
23 Arc, LazyLock,
24 atomic::{AtomicBool, Ordering},
25 },
26 time::Duration,
27};
28
29use ahash::AHashMap;
30use anyhow::Context;
31use http::Method;
32use jiff::{Timestamp, civil::Date};
33use nautilus_core::{
34 AtomicMap, AtomicTime, UUID4, nanos::UnixNanos, string::secret::SecretString,
35 time::get_atomic_clock_realtime,
36};
37use nautilus_model::{
38 data::{Bar, BookOrder, FundingRateUpdate, TradeTick},
39 enums::{BookType, OrderSide, OrderStatus, OrderType, TimeInForce},
40 events::AccountState,
41 fees::MakerTakerFeeRates,
42 identifiers::{AccountId, ClientOrderId, InstrumentId, VenueOrderId},
43 instruments::{Instrument, any::InstrumentAny},
44 orderbook::OrderBook,
45 reports::{FillReport, OrderStatusReport, PositionStatusReport},
46 types::{Price, Quantity},
47};
48use nautilus_network::{
49 http::{HttpClient, HttpRedirectPolicy, create_standard_nautilus_headers},
50 ratelimiter::quota::Quota,
51 retry::{RetryConfig, RetryError, RetryManager},
52};
53use parking_lot::RwLock;
54use rust_decimal::Decimal;
55use serde::{Serialize, de::DeserializeOwned};
56use tokio_util::sync::CancellationToken;
57use ustr::Ustr;
58
59use super::{
60 error::AxHttpError,
61 models::{
62 AuthenticateApiKeyRequest, AxAuthenticateResponse, AxBalancesResponse, AxBookResponse,
63 AxCancelAllOrdersResponse, AxCancelOrderResponse, AxCandle, AxCandleResponse,
64 AxCandlesResponse, AxFillsResponse, AxFundingRatesResponse, AxFundingSlotsResponse,
65 AxInitialMarginRequirementResponse, AxInstrument, AxInstrumentsResponse,
66 AxOpenOrdersResponse, AxOrderRejectReason, AxOrderStatusQueryResponse, AxOrdersResponse,
67 AxPlaceOrderResponse, AxPositionsResponse, AxPreviewAggressiveLimitOrderResponse,
68 AxReplaceOrderResponse, AxRiskSnapshotResponse, AxTicker, AxTickerResponse,
69 AxTickersResponse, AxTradesResponse, AxTransactionsResponse, AxWhoAmI,
70 CancelAllOrdersRequest, CancelOrderRequest, PlaceOrderRequest,
71 PreviewAggressiveLimitOrderRequest, ReplaceOrderRequest,
72 },
73 parse::{
74 parse_account_state, parse_bar, parse_fill_report, parse_funding_rate, parse_instrument,
75 parse_order_detail_status_report, parse_order_status, parse_order_status_report,
76 parse_position_status_report, parse_trade_tick,
77 },
78 query::{
79 GetBookParams, GetCandleParams, GetCandlesParams, GetFillsParams, GetFundingRatesParams,
80 GetFundingSlotsParams, GetInstrumentParams, GetOpenOrdersParams, GetOrderStatusParams,
81 GetOrdersParams, GetTickerParams, GetTickersParams, GetTradesParams, GetTransactionsParams,
82 },
83};
84use crate::common::{
85 consts::{AX_FILLS_MAX_LOOKBACK_DAYS, AX_HTTP_URL, AX_ORDERS_URL},
86 credential::Credential,
87 enums::{AxCandleWidth, AxInstrumentState},
88 parse::{ax_timestamp_stn_to_unix_nanos, cid_to_client_order_id, client_order_id_to_cid},
89};
90
91pub static AX_REST_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
95 Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant")
96});
97
98const AX_GLOBAL_RATE_KEY: &str = "architect:global";
99
100pub struct AxRawHttpClient {
105 base_url: String,
106 orders_base_url: String,
107 client: HttpClient,
108 credential: Option<Credential>,
109 session_token: RwLock<Option<SecretString>>,
110 retry_manager: RetryManager<AxHttpError>,
111 cancellation_token: RwLock<CancellationToken>,
112}
113
114impl Default for AxRawHttpClient {
115 fn default() -> Self {
116 Self::new(None, None, 60, 3, 1000, 10_000, None)
117 .expect("Failed to create default AxRawHttpClient")
118 }
119}
120
121impl Debug for AxRawHttpClient {
122 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
123 let has_session_token = self.session_token.read().is_some();
124 f.debug_struct(stringify!(AxRawHttpClient))
125 .field("base_url", &self.base_url)
126 .field("orders_base_url", &self.orders_base_url)
127 .field("has_credentials", &self.credential.is_some())
128 .field("has_session_token", &has_session_token)
129 .finish()
130 }
131}
132
133impl AxRawHttpClient {
134 #[must_use]
136 pub fn base_url(&self) -> &str {
137 &self.base_url
138 }
139
140 #[must_use]
142 pub fn api_key_masked(&self) -> String {
143 self.credential
144 .as_ref()
145 .map_or_else(|| "None".to_string(), |c| c.masked_api_key())
146 }
147
148 pub fn cancel_all_requests(&self) {
150 self.cancellation_token.read().cancel();
151 }
152
153 pub fn reset_cancellation_token(&self) {
155 *self.cancellation_token.write() = CancellationToken::new();
156 }
157
158 pub fn cancellation_token(&self) -> CancellationToken {
160 self.cancellation_token.read().clone()
161 }
162
163 pub fn new(
169 base_url: Option<String>,
170 orders_base_url: Option<String>,
171 timeout_secs: u64,
172 max_retries: u32,
173 retry_delay_ms: u64,
174 retry_delay_max_ms: u64,
175 proxy_url: Option<String>,
176 ) -> Result<Self, AxHttpError> {
177 let retry_config = RetryConfig {
178 max_retries,
179 initial_delay_ms: retry_delay_ms,
180 max_delay_ms: retry_delay_max_ms,
181 backoff_factor: 2.0,
182 jitter_ms: 1000,
183 operation_timeout_ms: Some(60_000),
184 immediate_first: false,
185 max_elapsed_ms: Some(180_000),
186 };
187
188 let retry_manager = RetryManager::new(retry_config);
189
190 Ok(Self {
191 base_url: base_url.unwrap_or_else(|| AX_HTTP_URL.to_string()),
192 orders_base_url: orders_base_url.unwrap_or_else(|| AX_ORDERS_URL.to_string()),
193 client: HttpClient::builder()
194 .redirect_policy(HttpRedirectPolicy::Reject)
195 .headers(Self::default_headers())
196 .keyed_quotas(Self::rate_limiter_quotas())
197 .default_quota(*AX_REST_QUOTA)
198 .timeout_secs(timeout_secs)
199 .maybe_proxy_url(proxy_url)
200 .build()
201 .map_err(|e| {
202 AxHttpError::NetworkError(format!("Failed to create HTTP client: {e}"))
203 })?,
204 credential: None,
205 session_token: RwLock::new(None),
206 retry_manager,
207 cancellation_token: RwLock::new(CancellationToken::new()),
208 })
209 }
210
211 #[expect(clippy::too_many_arguments)]
217 pub fn with_credentials(
218 api_key: String,
219 api_secret: String,
220 base_url: Option<String>,
221 orders_base_url: Option<String>,
222 timeout_secs: u64,
223 max_retries: u32,
224 retry_delay_ms: u64,
225 retry_delay_max_ms: u64,
226 proxy_url: Option<String>,
227 ) -> Result<Self, AxHttpError> {
228 let retry_config = RetryConfig {
229 max_retries,
230 initial_delay_ms: retry_delay_ms,
231 max_delay_ms: retry_delay_max_ms,
232 backoff_factor: 2.0,
233 jitter_ms: 1000,
234 operation_timeout_ms: Some(60_000),
235 immediate_first: false,
236 max_elapsed_ms: Some(180_000),
237 };
238
239 let retry_manager = RetryManager::new(retry_config);
240
241 Ok(Self {
242 base_url: base_url.unwrap_or_else(|| AX_HTTP_URL.to_string()),
243 orders_base_url: orders_base_url.unwrap_or_else(|| AX_ORDERS_URL.to_string()),
244 client: HttpClient::builder()
245 .redirect_policy(HttpRedirectPolicy::Reject)
246 .headers(Self::default_headers())
247 .keyed_quotas(Self::rate_limiter_quotas())
248 .default_quota(*AX_REST_QUOTA)
249 .timeout_secs(timeout_secs)
250 .maybe_proxy_url(proxy_url)
251 .build()
252 .map_err(|e| {
253 AxHttpError::NetworkError(format!("Failed to create HTTP client: {e}"))
254 })?,
255 credential: Some(Credential::new(api_key, api_secret)),
256 session_token: RwLock::new(None),
257 retry_manager,
258 cancellation_token: RwLock::new(CancellationToken::new()),
259 })
260 }
261
262 pub fn set_session_token(&self, token: SecretString) {
266 *self.session_token.write() = Some(token);
267 }
268
269 fn default_headers() -> HashMap<String, String> {
270 let mut headers: HashMap<String, String> =
271 create_standard_nautilus_headers().into_iter().collect();
272 headers.insert("Accept".to_string(), "application/json".to_string());
273 headers
274 }
275
276 fn rate_limiter_quotas() -> Vec<(String, Quota)> {
277 vec![(AX_GLOBAL_RATE_KEY.to_string(), *AX_REST_QUOTA)]
278 }
279
280 fn rate_limit_keys(endpoint: &str) -> Vec<String> {
281 let normalized = endpoint.split('?').next().unwrap_or(endpoint);
282 let route = format!("architect:{normalized}");
283
284 vec![AX_GLOBAL_RATE_KEY.to_string(), route]
285 }
286
287 fn auth_headers(&self) -> Result<HashMap<String, String>, AxHttpError> {
288 let guard = self.session_token.read();
289 let session_token = guard.as_ref().ok_or(AxHttpError::MissingSessionToken)?;
290
291 let mut headers = HashMap::new();
292 headers.insert(
293 "Authorization".to_string(),
294 format!("Bearer {}", session_token.expose_secret()),
295 );
296
297 Ok(headers)
298 }
299
300 async fn send_request<T: DeserializeOwned, P: Serialize>(
301 &self,
302 method: Method,
303 endpoint: &str,
304 params: Option<&P>,
305 body: Option<Vec<u8>>,
306 authenticate: bool,
307 ) -> Result<T, AxHttpError> {
308 self.send_request_to_url(&self.base_url, method, endpoint, params, body, authenticate)
309 .await
310 }
311
312 async fn send_request_to_url<T: DeserializeOwned, P: Serialize>(
313 &self,
314 base_url: &str,
315 method: Method,
316 endpoint: &str,
317 params: Option<&P>,
318 body: Option<Vec<u8>>,
319 authenticate: bool,
320 ) -> Result<T, AxHttpError> {
321 let endpoint = endpoint.to_string();
322 let url = format!("{base_url}{endpoint}");
323
324 let params_str = if method == Method::GET || method == Method::DELETE {
325 params
326 .map(serde_urlencoded::to_string)
327 .transpose()
328 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize params: {e}")))?
329 } else {
330 None
331 };
332
333 let operation = || {
334 let url = url.clone();
335 let method = method.clone();
336 let endpoint = endpoint.clone();
337 let params_str = params_str.clone();
338 let body = body.clone();
339
340 async move {
341 let mut headers = Self::default_headers();
342
343 if authenticate {
344 let auth_headers = self.auth_headers()?;
345 headers.extend(auth_headers);
346 }
347
348 if body.is_some() {
349 headers.insert("Content-Type".to_string(), "application/json".to_string());
350 }
351
352 let full_url = if let Some(ref query) = params_str {
353 if query.is_empty() {
354 url
355 } else {
356 format!("{url}?{query}")
357 }
358 } else {
359 url
360 };
361
362 let rate_limit_keys = Self::rate_limit_keys(&endpoint);
363
364 let response = self
365 .client
366 .request(
367 method,
368 full_url,
369 None,
370 Some(headers),
371 body,
372 None,
373 Some(rate_limit_keys),
374 )
375 .await?;
376
377 let status = response.status;
378 let response_body = String::from_utf8_lossy(&response.body).to_string();
379
380 if !status.is_success() {
381 return Err(AxHttpError::UnexpectedStatus {
382 status: status.as_u16(),
383 body: response_body,
384 });
385 }
386
387 serde_json::from_str(&response_body).map_err(|e| {
388 AxHttpError::JsonError(format!(
389 "Failed to deserialize response: {e}\nBody: {response_body}"
390 ))
391 })
392 }
393 };
394
395 let is_idempotent = matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
397 let should_retry = |error: &AxHttpError| -> bool { is_idempotent && error.is_retryable() };
398
399 let create_error = |error: RetryError| -> AxHttpError {
400 match error {
401 RetryError::Canceled => {
402 AxHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
403 }
404 error => AxHttpError::NetworkError(error.to_string()),
405 }
406 };
407
408 let cancel_token = self.cancellation_token.read().clone();
409
410 self.retry_manager
411 .invocation(endpoint.as_str(), operation, should_retry, create_error)
412 .cancellation_token(&cancel_token)
413 .execute()
414 .await
415 }
416
417 pub async fn get_whoami(&self) -> Result<AxWhoAmI, AxHttpError> {
426 self.send_request::<AxWhoAmI, ()>(Method::GET, "/whoami", None, None, true)
427 .await
428 }
429
430 pub async fn get_instruments(&self) -> Result<AxInstrumentsResponse, AxHttpError> {
439 self.send_request::<AxInstrumentsResponse, ()>(
440 Method::GET,
441 "/instruments",
442 None,
443 None,
444 false,
445 )
446 .await
447 }
448
449 pub async fn get_balances(&self) -> Result<AxBalancesResponse, AxHttpError> {
458 self.send_request::<AxBalancesResponse, ()>(Method::GET, "/balances", None, None, true)
459 .await
460 }
461
462 pub async fn get_positions(&self) -> Result<AxPositionsResponse, AxHttpError> {
471 self.send_request::<AxPositionsResponse, ()>(Method::GET, "/positions", None, None, true)
472 .await
473 }
474
475 pub async fn get_tickers(&self) -> Result<AxTickersResponse, AxHttpError> {
484 self.send_request::<AxTickersResponse, ()>(Method::GET, "/tickers", None, None, true)
485 .await
486 }
487
488 pub async fn get_tickers_with_params(
497 &self,
498 params: &GetTickersParams,
499 ) -> Result<AxTickersResponse, AxHttpError> {
500 self.send_request::<AxTickersResponse, _>(Method::GET, "/tickers", Some(params), None, true)
501 .await
502 }
503
504 pub async fn get_ticker(&self, symbol: Ustr) -> Result<AxTicker, AxHttpError> {
513 let params = GetTickerParams::new(symbol);
514 self.send_request::<AxTickerResponse, _>(Method::GET, "/ticker", Some(¶ms), None, true)
515 .await
516 .map(|response| response.ticker)
517 }
518
519 pub async fn get_instrument(&self, symbol: Ustr) -> Result<AxInstrument, AxHttpError> {
528 let params = GetInstrumentParams::new(symbol);
529 self.send_request::<AxInstrument, _>(Method::GET, "/instrument", Some(¶ms), None, false)
530 .await
531 }
532
533 pub async fn authenticate(
542 &self,
543 api_key: &str,
544 api_secret: &str,
545 expiration_seconds: i32,
546 ) -> Result<AxAuthenticateResponse, AxHttpError> {
547 let request = AuthenticateApiKeyRequest::new(api_key, api_secret, expiration_seconds);
548
549 let body = serde_json::to_vec(&request)
550 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
551
552 self.send_request::<AxAuthenticateResponse, ()>(
553 Method::POST,
554 "/authenticate",
555 None,
556 Some(body),
557 false,
558 )
559 .await
560 }
561
562 pub async fn authenticate_auto(
577 &self,
578 expiration_seconds: i32,
579 ) -> Result<AxAuthenticateResponse, AxHttpError> {
580 let (api_key, api_secret) = self
581 .resolve_credentials()
582 .ok_or(AxHttpError::MissingCredentials)?;
583
584 self.authenticate(
585 api_key.expose_secret(),
586 api_secret.expose_secret(),
587 expiration_seconds,
588 )
589 .await
590 }
591
592 fn resolve_credentials(&self) -> Option<(SecretString, SecretString)> {
593 if let Some(cred) = &self.credential {
594 return Some((
595 SecretString::from(cred.api_key()),
596 SecretString::from(cred.api_secret()),
597 ));
598 }
599
600 let cred = Credential::resolve(None, None)?;
601 Some((
602 SecretString::from(cred.api_key()),
603 SecretString::from(cred.api_secret()),
604 ))
605 }
606
607 pub async fn place_order(
616 &self,
617 request: &PlaceOrderRequest,
618 ) -> Result<AxPlaceOrderResponse, AxHttpError> {
619 let body = serde_json::to_vec(request)
620 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
621 self.send_request_to_url::<AxPlaceOrderResponse, ()>(
622 &self.orders_base_url,
623 Method::POST,
624 "/place-order",
625 None,
626 Some(body),
627 true,
628 )
629 .await
630 }
631
632 pub async fn cancel_order(&self, order_id: &str) -> Result<AxCancelOrderResponse, AxHttpError> {
641 let request = CancelOrderRequest::new(order_id);
642 let body = serde_json::to_vec(&request)
643 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
644 self.send_request_to_url::<AxCancelOrderResponse, ()>(
645 &self.orders_base_url,
646 Method::POST,
647 "/cancel-order",
648 None,
649 Some(body),
650 true,
651 )
652 .await
653 }
654
655 pub async fn replace_order(
667 &self,
668 request: &ReplaceOrderRequest,
669 ) -> Result<AxReplaceOrderResponse, AxHttpError> {
670 let body = serde_json::to_vec(request)
671 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
672 self.send_request_to_url::<AxReplaceOrderResponse, ()>(
673 &self.orders_base_url,
674 Method::POST,
675 "/replace-order",
676 None,
677 Some(body),
678 true,
679 )
680 .await
681 }
682
683 pub async fn cancel_all_orders(
692 &self,
693 request: &CancelAllOrdersRequest,
694 ) -> Result<AxCancelAllOrdersResponse, AxHttpError> {
695 let body = serde_json::to_vec(request)
696 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
697 self.send_request_to_url::<AxCancelAllOrdersResponse, ()>(
698 &self.orders_base_url,
699 Method::POST,
700 "/cancel-all-orders",
701 None,
702 Some(body),
703 true,
704 )
705 .await
706 }
707
708 pub async fn get_open_orders(&self) -> Result<AxOpenOrdersResponse, AxHttpError> {
717 self.get_open_orders_page(&GetOpenOrdersParams::new()).await
718 }
719
720 pub async fn get_open_orders_page(
726 &self,
727 params: &GetOpenOrdersParams,
728 ) -> Result<AxOpenOrdersResponse, AxHttpError> {
729 self.send_request_to_url::<AxOpenOrdersResponse, _>(
730 &self.orders_base_url,
731 Method::GET,
732 "/open-orders",
733 Some(params),
734 None,
735 true,
736 )
737 .await
738 }
739
740 pub async fn get_fills(
749 &self,
750 start_timestamp_ns: i64,
751 end_timestamp_ns: i64,
752 ) -> Result<AxFillsResponse, AxHttpError> {
753 let params = GetFillsParams::new(start_timestamp_ns, end_timestamp_ns);
754 self.get_fills_page(¶ms).await
755 }
756
757 pub async fn get_fills_page(
763 &self,
764 params: &GetFillsParams,
765 ) -> Result<AxFillsResponse, AxHttpError> {
766 self.send_request::<AxFillsResponse, _>(Method::GET, "/fills", Some(params), None, true)
767 .await
768 }
769
770 pub async fn get_candles(
779 &self,
780 symbol: Ustr,
781 start_timestamp_ns: i64,
782 end_timestamp_ns: i64,
783 candle_width: AxCandleWidth,
784 ) -> Result<AxCandlesResponse, AxHttpError> {
785 let params =
786 GetCandlesParams::new(symbol, start_timestamp_ns, end_timestamp_ns, candle_width);
787 self.send_request::<AxCandlesResponse, _>(
788 Method::GET,
789 "/candles",
790 Some(¶ms),
791 None,
792 true,
793 )
794 .await
795 }
796
797 pub async fn get_current_candle(
806 &self,
807 symbol: Ustr,
808 candle_width: AxCandleWidth,
809 ) -> Result<AxCandle, AxHttpError> {
810 let params = GetCandleParams::new(symbol, candle_width);
811 let response = self
812 .send_request::<AxCandleResponse, _>(
813 Method::GET,
814 "/candles/current",
815 Some(¶ms),
816 None,
817 true,
818 )
819 .await?;
820 Ok(response.candle)
821 }
822
823 pub async fn get_last_candle(
832 &self,
833 symbol: Ustr,
834 candle_width: AxCandleWidth,
835 ) -> Result<AxCandle, AxHttpError> {
836 let params = GetCandleParams::new(symbol, candle_width);
837 let response = self
838 .send_request::<AxCandleResponse, _>(
839 Method::GET,
840 "/candles/last",
841 Some(¶ms),
842 None,
843 true,
844 )
845 .await?;
846 Ok(response.candle)
847 }
848
849 pub async fn get_funding_rates(
858 &self,
859 symbol: Ustr,
860 start_timestamp_ns: i64,
861 end_timestamp_ns: i64,
862 ) -> Result<AxFundingRatesResponse, AxHttpError> {
863 let params = GetFundingRatesParams::new(symbol, start_timestamp_ns, end_timestamp_ns);
864 self.get_funding_rates_page(¶ms).await
865 }
866
867 pub async fn get_funding_rates_page(
873 &self,
874 params: &GetFundingRatesParams,
875 ) -> Result<AxFundingRatesResponse, AxHttpError> {
876 self.send_request::<AxFundingRatesResponse, _>(
877 Method::GET,
878 "/funding-rates",
879 Some(params),
880 None,
881 true,
882 )
883 .await
884 }
885
886 pub async fn get_funding_slots(
895 &self,
896 params: &GetFundingSlotsParams,
897 ) -> Result<AxFundingSlotsResponse, AxHttpError> {
898 self.send_request::<AxFundingSlotsResponse, _>(
899 Method::GET,
900 "/funding-slots",
901 Some(params),
902 None,
903 true,
904 )
905 .await
906 }
907
908 pub async fn get_risk_snapshot(&self) -> Result<AxRiskSnapshotResponse, AxHttpError> {
917 self.send_request::<AxRiskSnapshotResponse, ()>(
918 Method::GET,
919 "/risk-snapshot",
920 None,
921 None,
922 true,
923 )
924 .await
925 }
926
927 pub async fn preview_aggressive_limit_order(
940 &self,
941 request: &PreviewAggressiveLimitOrderRequest,
942 ) -> Result<AxPreviewAggressiveLimitOrderResponse, AxHttpError> {
943 let body = serde_json::to_vec(request)
944 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
945 self.send_request::<AxPreviewAggressiveLimitOrderResponse, ()>(
946 Method::POST,
947 "/preview-aggressive-limit-order",
948 None,
949 Some(body),
950 true,
951 )
952 .await
953 }
954
955 pub async fn get_transactions(
964 &self,
965 transaction_types: Vec<Ustr>,
966 start_timestamp_ns: i64,
967 end_timestamp_ns: i64,
968 ) -> Result<AxTransactionsResponse, AxHttpError> {
969 let params =
970 GetTransactionsParams::new(transaction_types, start_timestamp_ns, end_timestamp_ns);
971 self.get_transactions_page(¶ms).await
972 }
973
974 pub async fn get_transactions_page(
980 &self,
981 params: &GetTransactionsParams,
982 ) -> Result<AxTransactionsResponse, AxHttpError> {
983 self.send_request::<AxTransactionsResponse, _>(
984 Method::GET,
985 "/transactions",
986 Some(params),
987 None,
988 true,
989 )
990 .await
991 }
992
993 pub async fn get_trades(
1002 &self,
1003 symbol: Ustr,
1004 limit: Option<i32>,
1005 ) -> Result<AxTradesResponse, AxHttpError> {
1006 let params = GetTradesParams::new(symbol, limit);
1007 self.send_request::<AxTradesResponse, _>(Method::GET, "/trades", Some(¶ms), None, true)
1008 .await
1009 }
1010
1011 pub async fn get_book(
1020 &self,
1021 symbol: Ustr,
1022 level: Option<i32>,
1023 ) -> Result<AxBookResponse, AxHttpError> {
1024 let params = GetBookParams::new(symbol, level);
1025 self.send_request::<AxBookResponse, _>(Method::GET, "/book", Some(¶ms), None, true)
1027 .await
1028 }
1029
1030 pub async fn get_order_status_by_id(
1039 &self,
1040 order_id: &str,
1041 ) -> Result<AxOrderStatusQueryResponse, AxHttpError> {
1042 let params = GetOrderStatusParams::by_order_id(order_id);
1043 self.send_request_to_url::<AxOrderStatusQueryResponse, _>(
1044 &self.orders_base_url,
1045 Method::GET,
1046 "/order-status",
1047 Some(¶ms),
1048 None,
1049 true,
1050 )
1051 .await
1052 }
1053
1054 pub async fn get_order_status_by_cid(
1063 &self,
1064 client_order_id: u64,
1065 ) -> Result<AxOrderStatusQueryResponse, AxHttpError> {
1066 let params = GetOrderStatusParams::by_client_order_id(client_order_id);
1067 self.send_request_to_url::<AxOrderStatusQueryResponse, _>(
1068 &self.orders_base_url,
1069 Method::GET,
1070 "/order-status",
1071 Some(¶ms),
1072 None,
1073 true,
1074 )
1075 .await
1076 }
1077
1078 pub async fn get_orders(
1087 &self,
1088 params: &GetOrdersParams,
1089 ) -> Result<AxOrdersResponse, AxHttpError> {
1090 self.send_request_to_url::<AxOrdersResponse, _>(
1091 &self.orders_base_url,
1092 Method::GET,
1093 "/orders",
1094 Some(params),
1095 None,
1096 true,
1097 )
1098 .await
1099 }
1100
1101 pub async fn check_initial_margin(
1110 &self,
1111 request: &PlaceOrderRequest,
1112 ) -> Result<AxInitialMarginRequirementResponse, AxHttpError> {
1113 let body = serde_json::to_vec(request)
1114 .map_err(|e| AxHttpError::JsonError(format!("Failed to serialize request: {e}")))?;
1115 self.send_request_to_url::<AxInitialMarginRequirementResponse, ()>(
1116 &self.orders_base_url,
1117 Method::POST,
1118 "/initial-margin-requirement",
1119 None,
1120 Some(body),
1121 true,
1122 )
1123 .await
1124 }
1125}
1126
1127#[derive(Debug)]
1132#[cfg_attr(
1133 feature = "python",
1134 pyo3::pyclass(module = "nautilus_trader.adapters.architect_ax", from_py_object)
1135)]
1136#[cfg_attr(
1137 feature = "python",
1138 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.architect_ax")
1139)]
1140pub struct AxHttpClient {
1141 pub(crate) inner: Arc<AxRawHttpClient>,
1142 pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
1143 pub(crate) account_fee_rates: Arc<AtomicMap<Option<Ustr>, MakerTakerFeeRates>>,
1144 clock: &'static AtomicTime,
1145 cache_initialized: Arc<AtomicBool>,
1146}
1147
1148impl Clone for AxHttpClient {
1149 fn clone(&self) -> Self {
1150 Self {
1151 inner: self.inner.clone(),
1152 instruments_cache: self.instruments_cache.clone(),
1153 account_fee_rates: self.account_fee_rates.clone(),
1154 clock: self.clock,
1155 cache_initialized: self.cache_initialized.clone(),
1156 }
1157 }
1158}
1159
1160impl Default for AxHttpClient {
1161 fn default() -> Self {
1162 Self::new(None, None, 60, 3, 1000, 10_000, None)
1163 .expect("Failed to create default AxHttpClient")
1164 }
1165}
1166
1167impl AxHttpClient {
1168 pub fn new(
1174 base_url: Option<String>,
1175 orders_base_url: Option<String>,
1176 timeout_secs: u64,
1177 max_retries: u32,
1178 retry_delay_ms: u64,
1179 retry_delay_max_ms: u64,
1180 proxy_url: Option<String>,
1181 ) -> Result<Self, AxHttpError> {
1182 Ok(Self {
1183 inner: Arc::new(AxRawHttpClient::new(
1184 base_url,
1185 orders_base_url,
1186 timeout_secs,
1187 max_retries,
1188 retry_delay_ms,
1189 retry_delay_max_ms,
1190 proxy_url,
1191 )?),
1192 instruments_cache: Arc::new(AtomicMap::new()),
1193 account_fee_rates: Arc::new(AtomicMap::new()),
1194 clock: get_atomic_clock_realtime(),
1195 cache_initialized: Arc::new(AtomicBool::new(false)),
1196 })
1197 }
1198
1199 #[expect(clippy::too_many_arguments)]
1205 pub fn with_credentials(
1206 api_key: String,
1207 api_secret: String,
1208 base_url: Option<String>,
1209 orders_base_url: Option<String>,
1210 timeout_secs: u64,
1211 max_retries: u32,
1212 retry_delay_ms: u64,
1213 retry_delay_max_ms: u64,
1214 proxy_url: Option<String>,
1215 ) -> Result<Self, AxHttpError> {
1216 Ok(Self {
1217 inner: Arc::new(AxRawHttpClient::with_credentials(
1218 api_key,
1219 api_secret,
1220 base_url,
1221 orders_base_url,
1222 timeout_secs,
1223 max_retries,
1224 retry_delay_ms,
1225 retry_delay_max_ms,
1226 proxy_url,
1227 )?),
1228 instruments_cache: Arc::new(AtomicMap::new()),
1229 account_fee_rates: Arc::new(AtomicMap::new()),
1230 clock: get_atomic_clock_realtime(),
1231 cache_initialized: Arc::new(AtomicBool::new(false)),
1232 })
1233 }
1234
1235 #[must_use]
1237 pub fn base_url(&self) -> &str {
1238 self.inner.base_url()
1239 }
1240
1241 #[must_use]
1243 pub fn api_key_masked(&self) -> String {
1244 self.inner.api_key_masked()
1245 }
1246
1247 pub fn cancel_all_requests(&self) {
1249 self.inner.cancel_all_requests();
1250 }
1251
1252 pub fn reset_cancellation_token(&self) {
1254 self.inner.reset_cancellation_token();
1255 }
1256
1257 pub fn set_session_token(&self, token: SecretString) {
1261 self.inner.set_session_token(token);
1262 }
1263
1264 fn generate_ts_init(&self) -> UnixNanos {
1266 self.clock.get_time_ns()
1267 }
1268
1269 #[must_use]
1273 pub fn is_initialized(&self) -> bool {
1274 self.cache_initialized.load(Ordering::Acquire)
1275 }
1276
1277 #[must_use]
1279 pub fn get_cached_symbols(&self) -> Vec<String> {
1280 self.instruments_cache
1281 .load()
1282 .keys()
1283 .map(|k| k.to_string())
1284 .collect()
1285 }
1286
1287 pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
1291 self.instruments_cache.rcu(|m| {
1292 for inst in instruments {
1293 m.insert(inst.raw_symbol().inner(), inst.clone());
1294 }
1295 });
1296 self.cache_initialized.store(true, Ordering::Release);
1297 }
1298
1299 pub fn cache_instrument(&self, instrument: InstrumentAny) {
1303 self.instruments_cache
1304 .insert(instrument.raw_symbol().inner(), instrument);
1305 self.cache_initialized.store(true, Ordering::Release);
1306 }
1307
1308 pub async fn authenticate(
1316 &self,
1317 api_key: &str,
1318 api_secret: &str,
1319 expiration_seconds: i32,
1320 ) -> Result<SecretString, AxHttpError> {
1321 let resp = self
1322 .inner
1323 .authenticate(api_key, api_secret, expiration_seconds)
1324 .await?;
1325 let token = resp.into_token();
1326 self.inner.set_session_token(token.clone());
1327 Ok(token)
1328 }
1329
1330 pub async fn authenticate_auto(
1347 &self,
1348 expiration_seconds: i32,
1349 ) -> Result<SecretString, AxHttpError> {
1350 let resp = self.inner.authenticate_auto(expiration_seconds).await?;
1351 let token = resp.into_token();
1352 self.inner.set_session_token(token.clone());
1353 Ok(token)
1354 }
1355
1356 pub fn get_instrument(&self, symbol: &Ustr) -> Option<InstrumentAny> {
1358 self.instruments_cache.get_cloned(symbol)
1359 }
1360
1361 pub async fn request_account_fees(&self) -> anyhow::Result<(Decimal, Decimal)> {
1377 let whoami = self
1378 .inner
1379 .get_whoami()
1380 .await
1381 .map_err(|e| anyhow::anyhow!(e))
1382 .context("failed to request AX whoami")?;
1383
1384 let Some(account) = whoami.accounts.first() else {
1385 anyhow::bail!("AX whoami returned no accounts to resolve fees from");
1386 };
1387
1388 if whoami.accounts.len() > 1 {
1389 log::warn!(
1390 "AX credentials cover {} accounts, using {} as the fallback for fills without an account ID",
1391 whoami.accounts.len(),
1392 account.id,
1393 );
1394 }
1395
1396 let (Some(maker_fee), Some(taker_fee)) = (account.maker_fee, account.taker_fee) else {
1397 anyhow::bail!("AX whoami account {} supplied no fee rates", account.id);
1398 };
1399
1400 let fees = (maker_fee, taker_fee);
1401 let mut rates = AHashMap::new();
1402 rates.insert(None, MakerTakerFeeRates::new(maker_fee, taker_fee));
1403
1404 for account in &whoami.accounts {
1405 if let (Some(maker), Some(taker)) = (account.maker_fee, account.taker_fee) {
1406 rates.insert(Some(account.id), MakerTakerFeeRates::new(maker, taker));
1407 }
1408 }
1409
1410 self.account_fee_rates.store(rates);
1411
1412 Ok(fees)
1413 }
1414
1415 pub async fn request_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
1421 let resp = self
1422 .inner
1423 .get_instruments()
1424 .await
1425 .map_err(|e| anyhow::anyhow!(e))?;
1426
1427 let ts_init = self.generate_ts_init();
1428
1429 let mut instruments: Vec<InstrumentAny> = Vec::new();
1430 for inst in &resp.instruments {
1431 if inst.state == AxInstrumentState::Delisted {
1432 log::debug!("Skipping delisted instrument: {}", inst.symbol);
1433 continue;
1434 }
1435
1436 if inst.symbol.starts_with("TEST") {
1438 log::debug!("Skipping test instrument: {}", inst.symbol);
1439 continue;
1440 }
1441
1442 match parse_instrument(inst, ts_init, ts_init) {
1443 Ok(instrument) => instruments.push(instrument),
1444 Err(e) => {
1445 log::warn!("Failed to parse instrument {}: {e}", inst.symbol);
1446 }
1447 }
1448 }
1449
1450 Ok(instruments)
1451 }
1452
1453 pub async fn request_instrument(&self, symbol: Ustr) -> anyhow::Result<InstrumentAny> {
1459 let resp = self
1460 .inner
1461 .get_instrument(symbol)
1462 .await
1463 .map_err(|e| anyhow::anyhow!(e))?;
1464
1465 let ts_init = self.generate_ts_init();
1466
1467 parse_instrument(&resp, ts_init, ts_init)
1468 }
1469
1470 pub async fn request_book_snapshot(
1480 &self,
1481 symbol: Ustr,
1482 depth: Option<usize>,
1483 ) -> anyhow::Result<OrderBook> {
1484 let instrument = self
1485 .get_instrument(&symbol)
1486 .ok_or_else(|| anyhow::anyhow!("Instrument {symbol} not found in cache"))?;
1487
1488 let resp = self
1489 .inner
1490 .get_book(symbol, Some(2))
1491 .await
1492 .map_err(|e| anyhow::anyhow!(e))?;
1493
1494 let instrument_id = instrument.id();
1495 let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
1496
1497 let price_precision = instrument.price_precision();
1498 let size_precision = instrument.size_precision();
1499 let ts_event = ax_timestamp_stn_to_unix_nanos(resp.book.ts, resp.book.tn)?;
1500
1501 for (i, level) in resp.book.b.iter().enumerate() {
1502 if depth.is_some_and(|d| i >= d) {
1503 break;
1504 }
1505 let price = Price::from_decimal_dp(level.p, price_precision).with_context(|| {
1506 format!(
1507 "Failed to convert AX book bid price {} for {symbol}",
1508 level.p
1509 )
1510 })?;
1511 let size = Quantity::new(level.q as f64, size_precision);
1512 let order = BookOrder::new(OrderSide::Buy, price, size, i as u64);
1513 book.add(order, 0, i as u64, ts_event);
1514 }
1515
1516 let bids_len = resp.book.b.len();
1517 for (i, level) in resp.book.a.iter().enumerate() {
1518 if depth.is_some_and(|d| i >= d) {
1519 break;
1520 }
1521 let price = Price::from_decimal_dp(level.p, price_precision).with_context(|| {
1522 format!(
1523 "Failed to convert AX book ask price {} for {symbol}",
1524 level.p
1525 )
1526 })?;
1527 let size = Quantity::new(level.q as f64, size_precision);
1528 let order = BookOrder::new(OrderSide::Sell, price, size, (bids_len + i) as u64);
1529 book.add(order, 0, (bids_len + i) as u64, ts_event);
1530 }
1531
1532 Ok(book)
1533 }
1534
1535 pub async fn request_trade_ticks(
1549 &self,
1550 symbol: Ustr,
1551 limit: Option<i32>,
1552 start: Option<UnixNanos>,
1553 end: Option<UnixNanos>,
1554 ) -> anyhow::Result<Vec<TradeTick>> {
1555 let instrument = self
1556 .get_instrument(&symbol)
1557 .ok_or_else(|| anyhow::anyhow!("Instrument {symbol} not found in cache"))?;
1558
1559 let resp = self
1560 .inner
1561 .get_trades(symbol, limit)
1562 .await
1563 .map_err(|e| anyhow::anyhow!(e))?;
1564
1565 let ts_init = self.generate_ts_init();
1566 let mut ticks = Vec::with_capacity(resp.trades.len());
1567
1568 for trade in &resp.trades {
1569 match parse_trade_tick(trade, &instrument, ts_init) {
1570 Ok(tick) => {
1571 if start.is_some_and(|s| tick.ts_event < s) {
1572 continue;
1573 }
1574
1575 if end.is_some_and(|e| tick.ts_event > e) {
1576 continue;
1577 }
1578 ticks.push(tick);
1579 }
1580 Err(e) => {
1581 log::warn!("Failed to parse trade for {symbol}: {e}");
1582 }
1583 }
1584 }
1585
1586 Ok(ticks)
1587 }
1588
1589 pub async fn request_bars(
1600 &self,
1601 symbol: Ustr,
1602 start: Option<Timestamp>,
1603 end: Option<Timestamp>,
1604 width: AxCandleWidth,
1605 ) -> anyhow::Result<Vec<Bar>> {
1606 let instrument = self
1607 .get_instrument(&symbol)
1608 .ok_or_else(|| anyhow::anyhow!("Instrument {symbol} not found in cache"))?;
1609
1610 let start_ns = start
1611 .and_then(|dt| i64::try_from(dt.as_nanosecond()).ok())
1612 .unwrap_or(0);
1613 let end_ns = end
1614 .and_then(|dt| i64::try_from(dt.as_nanosecond()).ok())
1615 .unwrap_or_else(|| self.generate_ts_init().as_i64());
1616 let resp = self
1617 .inner
1618 .get_candles(symbol, start_ns, end_ns, width)
1619 .await
1620 .map_err(|e| anyhow::anyhow!(e))?;
1621
1622 let ts_init = self.generate_ts_init();
1623 let mut bars = Vec::with_capacity(resp.candles.len());
1624
1625 for candle in &resp.candles {
1626 match parse_bar(candle, &instrument, ts_init) {
1627 Ok(bar) => bars.push(bar),
1628 Err(e) => {
1629 log::warn!("Failed to parse bar for {symbol}: {e}");
1630 }
1631 }
1632 }
1633
1634 Ok(bars)
1635 }
1636
1637 pub async fn request_funding_rates(
1646 &self,
1647 instrument_id: InstrumentId,
1648 start: Option<Timestamp>,
1649 end: Option<Timestamp>,
1650 ) -> Result<Vec<FundingRateUpdate>, AxHttpError> {
1651 const PAGE_SIZE: i32 = 100;
1652
1653 let symbol = instrument_id.symbol.inner();
1654 let start_ns = start
1655 .and_then(|dt| i64::try_from(dt.as_nanosecond()).ok())
1656 .unwrap_or(0);
1657 let end_ns = end
1658 .and_then(|dt| i64::try_from(dt.as_nanosecond()).ok())
1659 .unwrap_or_else(|| self.generate_ts_init().as_i64());
1660 let mut params = GetFundingRatesParams::new(symbol, start_ns, end_ns);
1661 params.limit = Some(PAGE_SIZE);
1662 params.sort_ts = Some("desc".to_string());
1663
1664 let mut funding_rates = Vec::new();
1665 let mut seen_rows = HashSet::new();
1666 let mut seen_cursors = HashSet::new();
1667 let mut expected_total = None;
1668
1669 loop {
1670 let response = self.inner.get_funding_rates_page(¶ms).await?;
1671 let page_len = response.funding_rates.len();
1672
1673 if page_len > PAGE_SIZE as usize {
1674 return Err(format!(
1675 "AX funding-rates page length {page_len} exceeds requested limit {PAGE_SIZE}"
1676 )
1677 .into());
1678 }
1679
1680 if let Some(limit) = response.limit {
1681 if !(0..=PAGE_SIZE).contains(&limit) {
1682 return Err(format!(
1683 "AX funding-rates applied limit must be between 0 and {PAGE_SIZE}, was {limit}"
1684 )
1685 .into());
1686 }
1687
1688 if page_len > limit as usize {
1689 return Err(format!(
1690 "AX funding-rates page length {page_len} exceeds applied limit {limit}"
1691 )
1692 .into());
1693 }
1694 }
1695
1696 if let Some(total_count) = response.total_count {
1697 if total_count < 0 {
1698 return Err(format!(
1699 "AX funding-rates total_count must be non-negative, was {total_count}"
1700 )
1701 .into());
1702 }
1703
1704 if let Some(expected) = expected_total {
1705 if total_count != expected {
1706 return Err(format!(
1707 "AX funding-rates total_count changed during pagination: expected {expected}, was {total_count}"
1708 )
1709 .into());
1710 }
1711 } else {
1712 expected_total = Some(total_count);
1713 }
1714 }
1715
1716 for rate in response.funding_rates {
1717 let identity = (
1718 rate.symbol,
1719 rate.timestamp_ns,
1720 rate.funding_rate,
1721 rate.funding_amount,
1722 rate.benchmark_price,
1723 rate.settlement_price,
1724 );
1725
1726 if !seen_rows.insert(identity) {
1727 return Err(format!(
1728 "AX funding-rates pagination returned an exact duplicate row for {} at {}",
1729 rate.symbol, rate.timestamp_ns
1730 )
1731 .into());
1732 }
1733 funding_rates.push(rate);
1734 }
1735
1736 if let Some(total_count) = expected_total
1737 && funding_rates.len() as i64 > total_count
1738 {
1739 return Err(format!(
1740 "AX funding-rates pagination returned more unique rows ({}) than total_count {total_count}",
1741 funding_rates.len()
1742 )
1743 .into());
1744 }
1745
1746 match response.next_cursor {
1747 Some(next_cursor) => {
1748 if next_cursor.is_empty() {
1749 return Err("AX funding-rates returned an empty next_cursor"
1750 .to_string()
1751 .into());
1752 }
1753
1754 if page_len == 0 {
1755 return Err("AX funding-rates returned an empty page with a next_cursor"
1756 .to_string()
1757 .into());
1758 }
1759
1760 if !seen_cursors.insert(next_cursor.clone()) {
1761 return Err(format!(
1762 "AX funding-rates pagination repeated cursor {next_cursor:?}"
1763 )
1764 .into());
1765 }
1766 params.cursor = Some(next_cursor);
1767 }
1768 None => break,
1769 }
1770 }
1771
1772 if let Some(total_count) = expected_total
1773 && funding_rates.len() as i64 != total_count
1774 {
1775 return Err(format!(
1776 "AX funding-rates pagination returned {} unique rows, expected {total_count}",
1777 funding_rates.len()
1778 )
1779 .into());
1780 }
1781
1782 let ts_init = self.generate_ts_init();
1783 let updates = funding_rates
1784 .iter()
1785 .map(|r| parse_funding_rate(r, instrument_id, ts_init))
1786 .collect::<anyhow::Result<Vec<_>>>()
1787 .map_err(|e| AxHttpError::from(e.to_string()))?;
1788
1789 Ok(updates)
1790 }
1791
1792 pub async fn request_funding_slots(
1802 &self,
1803 instrument_id: InstrumentId,
1804 date: Option<Date>,
1805 ) -> Result<AxFundingSlotsResponse, AxHttpError> {
1806 let symbol = instrument_id.symbol.inner();
1807 let mut params = GetFundingSlotsParams::new(symbol);
1808
1809 if let Some(date) = date {
1810 params.date = Some(date.strftime("%Y-%m-%d").to_string());
1811 }
1812
1813 self.inner.get_funding_slots(¶ms).await
1814 }
1815
1816 pub async fn request_account_state(
1822 &self,
1823 account_id: AccountId,
1824 ) -> anyhow::Result<AccountState> {
1825 const RISK_SNAPSHOT_TIMEOUT_SECS: u64 = 10;
1827
1828 let (balances, risk) = tokio::join!(
1829 self.inner.get_balances(),
1830 tokio::time::timeout(
1831 Duration::from_secs(RISK_SNAPSHOT_TIMEOUT_SECS),
1832 self.inner.get_risk_snapshot(),
1833 ),
1834 );
1835
1836 let response = balances.map_err(|e| anyhow::anyhow!(e))?;
1837
1838 let risk = match risk {
1839 Ok(Ok(snapshot)) => Some(snapshot.risk_snapshot),
1840 Ok(Err(e)) => {
1841 log::warn!(
1842 "AX risk snapshot unavailable, account state reports zero locked margin: {e}"
1843 );
1844 None
1845 }
1846 Err(_) => {
1847 log::warn!(
1848 "AX risk snapshot timed out after {RISK_SNAPSHOT_TIMEOUT_SECS}s, \
1849 account state reports zero locked margin"
1850 );
1851 None
1852 }
1853 };
1854
1855 let ts_init = self.generate_ts_init();
1856 parse_account_state(&response, risk.as_ref(), account_id, ts_init, ts_init)
1857 }
1858
1859 pub async fn check_initial_margin(
1865 &self,
1866 request: &PlaceOrderRequest,
1867 ) -> anyhow::Result<Decimal> {
1868 let resp = self
1869 .inner
1870 .check_initial_margin(request)
1871 .await
1872 .map_err(|e| anyhow::anyhow!(e))?;
1873 Ok(resp.im)
1874 }
1875
1876 #[expect(clippy::too_many_arguments)]
1891 pub async fn request_order_status(
1892 &self,
1893 account_id: AccountId,
1894 instrument_id: InstrumentId,
1895 client_order_id: Option<ClientOrderId>,
1896 venue_order_id: Option<VenueOrderId>,
1897 order_side: Option<OrderSide>,
1898 order_type: OrderType,
1899 time_in_force: TimeInForce,
1900 ) -> anyhow::Result<OrderStatusReport> {
1901 let resp = if let Some(ref voi) = venue_order_id {
1902 self.inner.get_order_status_by_id(voi.as_str()).await
1903 } else if let Some(ref coid) = client_order_id {
1904 let cid = client_order_id_to_cid(coid);
1905 self.inner.get_order_status_by_cid(cid).await
1906 } else {
1907 anyhow::bail!("Either venue_order_id or client_order_id must be provided")
1908 }
1909 .map_err(|e| anyhow::anyhow!(e))?;
1910
1911 let detail = resp.status;
1912 let size_precision = self
1913 .get_instrument(&detail.symbol)
1914 .map_or(0, |i| i.size_precision());
1915
1916 let voi = VenueOrderId::new(&detail.order_id);
1917 let order_status = parse_order_status(detail.state, time_in_force)?;
1918 let filled = detail.filled_quantity.unwrap_or(0);
1919 let remaining = detail.remaining_quantity.unwrap_or(0);
1920
1921 let total = if remaining == 0
1922 && matches!(
1923 order_status,
1924 OrderStatus::Canceled | OrderStatus::Expired | OrderStatus::Rejected
1925 ) {
1926 let response = self
1927 .inner
1928 .get_orders(&GetOrdersParams {
1929 order_id: Some(detail.order_id.clone()),
1930 limit: Some(1),
1931 ..Default::default()
1932 })
1933 .await?;
1934
1935 let order = response
1936 .orders
1937 .iter()
1938 .find(|order| order.oid == detail.order_id)
1939 .with_context(|| {
1940 format!(
1941 "AX order {} original quantity unavailable in history",
1942 detail.order_id
1943 )
1944 })?;
1945
1946 Decimal::from(order.q)
1947 } else {
1948 Decimal::from(filled) + Decimal::from(remaining)
1949 };
1950
1951 let quantity = Quantity::from_decimal_dp(total, size_precision)?;
1952 let filled_qty = Quantity::from_decimal_dp(Decimal::from(filled), size_precision)?;
1953 let ts_init = self.generate_ts_init();
1954
1955 let resolved_coid = client_order_id.or_else(|| detail.clord_id.map(cid_to_client_order_id));
1956
1957 let mut report = OrderStatusReport::new(
1958 account_id,
1959 instrument_id,
1960 resolved_coid,
1961 voi,
1962 order_side,
1963 order_type,
1964 time_in_force,
1965 order_status,
1966 quantity,
1967 filled_qty,
1968 ts_init,
1969 ts_init,
1970 ts_init,
1971 Some(UUID4::new()),
1972 );
1973
1974 if let Some(reason) =
1975 AxOrderRejectReason::reason_str(detail.reject_reason, detail.reject_message.as_deref())
1976 {
1977 report = report.with_cancel_reason(reason);
1978 }
1979
1980 Ok(report)
1981 }
1982
1983 pub async fn request_order_status_reports<F>(
1997 &self,
1998 account_id: AccountId,
1999 cid_resolver: Option<F>,
2000 ) -> anyhow::Result<Vec<OrderStatusReport>>
2001 where
2002 F: Fn(u64) -> Option<ClientOrderId>,
2003 {
2004 const PAGE_SIZE: i32 = 100;
2005
2006 let mut orders = Vec::new();
2007 let mut seen_order_ids = HashSet::new();
2008 let mut offset = 0_i64;
2009 let mut expected_total = None;
2010
2011 loop {
2012 let request_offset = i32::try_from(offset)
2013 .context("AX open-orders offset exceeds the documented int32 range")?;
2014 let params = GetOpenOrdersParams {
2015 account_id: None,
2016 limit: Some(PAGE_SIZE),
2017 offset: Some(request_offset),
2018 sort_ts: Some("desc".to_string()),
2019 };
2020 let response = self
2021 .inner
2022 .get_open_orders_page(¶ms)
2023 .await
2024 .map_err(|e| anyhow::anyhow!(e))?;
2025
2026 anyhow::ensure!(
2027 response.total_count >= 0,
2028 "AX open-orders total_count must be non-negative, was {}",
2029 response.total_count
2030 );
2031 anyhow::ensure!(
2032 response.limit >= 0 && response.limit <= PAGE_SIZE,
2033 "AX open-orders applied limit must be between 0 and {PAGE_SIZE}, was {}",
2034 response.limit
2035 );
2036 anyhow::ensure!(
2037 i64::from(response.offset) == offset,
2038 "AX open-orders response offset mismatch: requested {offset}, was {}",
2039 response.offset
2040 );
2041
2042 let total_count = *expected_total.get_or_insert(response.total_count);
2043 anyhow::ensure!(
2044 response.total_count == total_count,
2045 "AX open-orders total_count changed during pagination: expected {total_count}, was {}",
2046 response.total_count
2047 );
2048
2049 let page_len = i64::try_from(response.orders.len())
2050 .context("AX open-orders page length exceeds i64")?;
2051 anyhow::ensure!(
2052 page_len <= i64::from(response.limit),
2053 "AX open-orders page length {page_len} exceeds applied limit {}",
2054 response.limit
2055 );
2056 let next_offset = offset
2057 .checked_add(page_len)
2058 .context("AX open-orders offset overflow")?;
2059 anyhow::ensure!(
2060 next_offset <= total_count,
2061 "AX open-orders page exceeds total_count: next offset {next_offset}, total {total_count}"
2062 );
2063
2064 if total_count == 0 {
2065 anyhow::ensure!(
2066 response.orders.is_empty(),
2067 "AX open-orders returned rows with total_count zero"
2068 );
2069 break;
2070 }
2071
2072 anyhow::ensure!(
2073 !response.orders.is_empty(),
2074 "AX open-orders returned an empty page before offset {offset} reached total {total_count}"
2075 );
2076
2077 for order in response.orders {
2078 anyhow::ensure!(
2079 seen_order_ids.insert(order.oid.clone()),
2080 "AX open-orders pagination returned duplicate order ID {}",
2081 order.oid
2082 );
2083 orders.push(order);
2084 }
2085
2086 if next_offset == total_count {
2087 break;
2088 }
2089
2090 offset = next_offset;
2091 }
2092
2093 anyhow::ensure!(
2094 i64::try_from(orders.len()).context("AX open-orders result length exceeds i64")?
2095 == expected_total.unwrap_or_default(),
2096 "AX open-orders pagination did not return the advertised number of unique orders"
2097 );
2098
2099 let ts_init = self.generate_ts_init();
2100 let mut reports = Vec::with_capacity(orders.len());
2101
2102 for order in &orders {
2103 let instrument = self.resolve_report_instrument(order.s).await?;
2104
2105 let report = parse_order_status_report(
2106 order,
2107 account_id,
2108 &instrument,
2109 ts_init,
2110 cid_resolver.as_ref(),
2111 )
2112 .with_context(|| format!("failed to parse AX order {}", order.oid))?;
2113 reports.push(report);
2114 }
2115
2116 Ok(reports)
2117 }
2118
2119 pub async fn request_historical_order_status_reports<F>(
2134 &self,
2135 account_id: AccountId,
2136 start: Option<UnixNanos>,
2137 end: Option<UnixNanos>,
2138 cid_resolver: Option<F>,
2139 ) -> anyhow::Result<Vec<OrderStatusReport>>
2140 where
2141 F: Fn(u64) -> Option<ClientOrderId>,
2142 {
2143 const PAGE_SIZE: i32 = 100;
2144
2145 let mut params = GetOrdersParams {
2146 start_timestamp_ns: start.map(|timestamp| timestamp.as_i64()),
2147 end_timestamp_ns: end.map(|timestamp| timestamp.as_i64()),
2148 limit: Some(PAGE_SIZE),
2149 ..Default::default()
2150 };
2151 let mut orders = Vec::new();
2152 let mut seen_cursors = HashSet::new();
2153 let mut seen_order_ids = HashSet::new();
2154
2155 loop {
2156 let response = self
2157 .inner
2158 .get_orders(¶ms)
2159 .await
2160 .map_err(|e| anyhow::anyhow!(e))?;
2161
2162 for order in response.orders {
2163 anyhow::ensure!(
2164 seen_order_ids.insert(order.oid.clone()),
2165 "AX orders pagination returned duplicate order ID {}",
2166 order.oid
2167 );
2168 orders.push(order);
2169 }
2170
2171 match response.next_cursor {
2172 Some(next_cursor) => {
2173 anyhow::ensure!(
2174 seen_cursors.insert(next_cursor.clone()),
2175 "AX orders pagination repeated cursor {next_cursor:?}"
2176 );
2177 params.cursor = Some(next_cursor);
2178 }
2179 None => break,
2180 }
2181 }
2182
2183 let ts_init = self.generate_ts_init();
2184 let mut reports = Vec::with_capacity(orders.len());
2185
2186 for order in &orders {
2187 let instrument = self.resolve_report_instrument(order.s).await?;
2188
2189 let report = parse_order_detail_status_report(
2190 order,
2191 account_id,
2192 &instrument,
2193 ts_init,
2194 cid_resolver.as_ref(),
2195 )
2196 .with_context(|| format!("failed to parse AX order {}", order.oid))?;
2197 reports.push(report);
2198 }
2199
2200 Ok(reports)
2201 }
2202
2203 pub async fn request_fill_reports(
2216 &self,
2217 account_id: AccountId,
2218 start: Option<UnixNanos>,
2219 end: Option<UnixNanos>,
2220 ) -> anyhow::Result<Vec<FillReport>> {
2221 const PAGE_SIZE: i32 = 100;
2222
2223 let max_span_ns = AX_FILLS_MAX_LOOKBACK_DAYS * 24 * 60 * 60 * 1_000_000_000;
2225 let end_ns = end.map_or_else(|| self.generate_ts_init().as_i64(), |e| e.as_i64());
2226 let floor_ns = end_ns - max_span_ns;
2227 let start_ns = start.map_or(floor_ns, |s| s.as_i64().max(floor_ns));
2228 let mut params = GetFillsParams::new(start_ns, end_ns);
2229 params.limit = Some(PAGE_SIZE);
2230 params.sort_ts = Some("desc".to_string());
2231
2232 let mut fills = Vec::new();
2233 let mut seen_trade_ids = HashSet::new();
2234 let mut seen_cursors = HashSet::new();
2235 let mut expected_total = None;
2236
2237 loop {
2238 let response = self
2239 .inner
2240 .get_fills_page(¶ms)
2241 .await
2242 .map_err(|e| anyhow::anyhow!(e))?;
2243 let page_len = response.fills.len();
2244
2245 anyhow::ensure!(
2246 page_len <= PAGE_SIZE as usize,
2247 "AX fills page length {page_len} exceeds requested limit {PAGE_SIZE}"
2248 );
2249
2250 if let Some(limit) = response.limit {
2251 anyhow::ensure!(
2252 (0..=PAGE_SIZE).contains(&limit),
2253 "AX fills applied limit must be between 0 and {PAGE_SIZE}, was {limit}"
2254 );
2255 anyhow::ensure!(
2256 page_len <= limit as usize,
2257 "AX fills page length {page_len} exceeds applied limit {limit}"
2258 );
2259 }
2260
2261 if let Some(total_count) = response.total_count {
2262 anyhow::ensure!(
2263 total_count >= 0,
2264 "AX fills total_count must be non-negative, was {total_count}"
2265 );
2266
2267 if let Some(expected) = expected_total {
2268 anyhow::ensure!(
2269 total_count == expected,
2270 "AX fills total_count changed during pagination: expected {expected}, was {total_count}"
2271 );
2272 } else {
2273 expected_total = Some(total_count);
2274 }
2275 }
2276
2277 for fill in response.fills {
2278 anyhow::ensure!(
2279 seen_trade_ids.insert(fill.trade_id.clone()),
2280 "AX fills pagination returned duplicate trade ID {}",
2281 fill.trade_id
2282 );
2283 fills.push(fill);
2284 }
2285
2286 if let Some(total_count) = expected_total {
2287 anyhow::ensure!(
2288 fills.len() as i64 <= total_count,
2289 "AX fills pagination returned more unique rows ({}) than total_count {total_count}",
2290 fills.len()
2291 );
2292 }
2293
2294 match response.next_cursor {
2295 Some(next_cursor) => {
2296 anyhow::ensure!(
2297 !next_cursor.is_empty(),
2298 "AX fills returned an empty next_cursor"
2299 );
2300 anyhow::ensure!(
2301 page_len > 0,
2302 "AX fills returned an empty page with a next_cursor"
2303 );
2304 anyhow::ensure!(
2305 seen_cursors.insert(next_cursor.clone()),
2306 "AX fills pagination repeated cursor {next_cursor:?}"
2307 );
2308 params.cursor = Some(next_cursor);
2309 }
2310 None => break,
2311 }
2312 }
2313
2314 if let Some(total_count) = expected_total {
2315 anyhow::ensure!(
2316 fills.len() as i64 == total_count,
2317 "AX fills pagination returned {} unique rows, expected {total_count}",
2318 fills.len()
2319 );
2320 }
2321
2322 let ts_init = self.generate_ts_init();
2323 let mut reports = Vec::with_capacity(fills.len());
2324
2325 for fill in &fills {
2326 let instrument = self.resolve_report_instrument(fill.symbol).await?;
2327 let report = parse_fill_report(fill, account_id, &instrument, ts_init)
2328 .with_context(|| format!("Failed to parse AX fill {}", fill.trade_id))?;
2329 reports.push(report);
2330 }
2331
2332 Ok(reports)
2333 }
2334
2335 pub async fn request_position_reports(
2349 &self,
2350 account_id: AccountId,
2351 ) -> anyhow::Result<Vec<PositionStatusReport>> {
2352 let response = self
2353 .inner
2354 .get_positions()
2355 .await
2356 .map_err(|e| anyhow::anyhow!(e))?;
2357
2358 let ts_init = self.generate_ts_init();
2359 let mut reports = Vec::with_capacity(response.positions.len());
2360
2361 for position in &response.positions {
2362 if position.signed_quantity == 0 {
2364 continue;
2365 }
2366
2367 let instrument = self.resolve_report_instrument(position.symbol).await?;
2368
2369 match parse_position_status_report(position, account_id, &instrument, ts_init) {
2370 Ok(report) => reports.push(report),
2371 Err(e) => {
2372 log::warn!("Failed to parse position for {}: {e}", position.symbol);
2373 }
2374 }
2375 }
2376
2377 Ok(reports)
2378 }
2379
2380 async fn resolve_report_instrument(&self, symbol: Ustr) -> anyhow::Result<InstrumentAny> {
2381 if let Some(instrument) = self.get_instrument(&symbol) {
2382 return Ok(instrument);
2383 }
2384
2385 let instrument = self.request_instrument(symbol).await.map_err(|e| {
2386 anyhow::anyhow!("Failed to resolve AX instrument {symbol} via GET /instrument: {e}")
2387 })?;
2388 self.cache_instrument(instrument.clone());
2389 Ok(instrument)
2390 }
2391
2392 pub async fn cancel_all_orders(&self, instrument_id: InstrumentId) -> Result<(), AxHttpError> {
2398 let request = CancelAllOrdersRequest::new().with_symbol(instrument_id.symbol.inner());
2399 self.inner.cancel_all_orders(&request).await?;
2400 Ok(())
2401 }
2402}
2403
2404#[cfg(test)]
2405mod tests {
2406 use nautilus_testkit::http::assert_http_redirect_rejected;
2407 use rstest::rstest;
2408
2409 use super::*;
2410
2411 #[rstest]
2412 #[tokio::test]
2413 async fn test_request_account_fees_caches_accounts_and_replaces_rates() {
2414 let mut whoami: AxWhoAmI =
2415 serde_json::from_str(include_str!("../../test_data/http_get_whoami.json")).unwrap();
2416 let mut secondary = whoami.accounts[0].clone();
2417 secondary.id = Ustr::from("secondary");
2418 secondary.maker_fee = Some(Decimal::new(-1, 4));
2419 secondary.taker_fee = Some(Decimal::new(8, 4));
2420 whoami.accounts.push(secondary);
2421 let mut incomplete = whoami.accounts[0].clone();
2422 incomplete.id = Ustr::from("incomplete");
2423 incomplete.taker_fee = None;
2424 whoami.accounts.push(incomplete);
2425 let response = Arc::new(tokio::sync::Mutex::new(whoami));
2426 let state = response.clone();
2427 let app = axum::Router::new().route(
2428 "/whoami",
2429 axum::routing::get(move || {
2430 let state = state.clone();
2431 async move { axum::Json(state.lock().await.clone()) }
2432 }),
2433 );
2434
2435 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2436 let addr = listener.local_addr().unwrap();
2437 let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
2438 let client =
2439 AxHttpClient::new(Some(format!("http://{addr}")), None, 3, 0, 1, 1, None).unwrap();
2440 client.set_session_token("test_session_token".into());
2441 let cloned = client.clone();
2442
2443 let original = client.request_account_fees().await.unwrap();
2444 let original_rates = cloned.account_fee_rates.load();
2445 let default_id = response.lock().await.accounts[0].id;
2446 let expected = AHashMap::from_iter([
2447 (
2448 None,
2449 MakerTakerFeeRates::new(Decimal::new(2, 4), Decimal::new(25, 4)),
2450 ),
2451 (
2452 Some(default_id),
2453 MakerTakerFeeRates::new(Decimal::new(2, 4), Decimal::new(25, 4)),
2454 ),
2455 (
2456 Some(Ustr::from("secondary")),
2457 MakerTakerFeeRates::new(Decimal::new(-1, 4), Decimal::new(8, 4)),
2458 ),
2459 ]);
2460 {
2461 let mut response = response.lock().await;
2462 response.accounts.truncate(1);
2463 response.accounts[0].maker_fee = Some(Decimal::ZERO);
2464 response.accounts[0].taker_fee = Some(Decimal::new(3, 3));
2465 }
2466
2467 let refreshed = cloned.request_account_fees().await.unwrap();
2468 let refreshed_rates = client.account_fee_rates.load();
2469 let expected_refreshed = AHashMap::from_iter([
2470 (
2471 None,
2472 MakerTakerFeeRates::new(Decimal::ZERO, Decimal::new(3, 3)),
2473 ),
2474 (
2475 Some(default_id),
2476 MakerTakerFeeRates::new(Decimal::ZERO, Decimal::new(3, 3)),
2477 ),
2478 ]);
2479
2480 assert_eq!(original, (Decimal::new(2, 4), Decimal::new(25, 4)));
2481 assert_eq!(**original_rates, expected);
2482 assert_eq!(refreshed, (Decimal::ZERO, Decimal::new(3, 3)));
2483 assert_eq!(**refreshed_rates, expected_refreshed);
2484 server.abort();
2485 }
2486
2487 #[rstest]
2488 #[case::credentials(true)]
2489 #[case::session_token(false)]
2490 #[tokio::test]
2491 async fn test_authenticated_client_rejects_redirects(#[case] credentials: bool) {
2492 let client = if credentials {
2493 AxRawHttpClient::with_credentials(
2494 "key".into(),
2495 "secret".into(),
2496 None,
2497 None,
2498 3,
2499 0,
2500 1,
2501 1,
2502 None,
2503 )
2504 .unwrap()
2505 } else {
2506 AxRawHttpClient::new(None, None, 3, 0, 1, 1, None).unwrap()
2507 }
2508 .client;
2509 assert_http_redirect_rejected(|url| async move {
2510 client
2511 .get(url, None, None, Some(3), None)
2512 .await
2513 .unwrap()
2514 .status
2515 .as_u16()
2516 })
2517 .await;
2518 }
2519}