Skip to main content

nautilus_architect_ax/http/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Provides the HTTP client integration for the Ax REST API.
17
18use 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
91/// Default Ax REST API rate limit.
92///
93/// Conservative default of 10 requests per second.
94pub 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
100/// Raw HTTP client for low-level AX Exchange API operations.
101///
102/// This client handles request/response operations with the AX Exchange API,
103/// returning venue-specific response types. It does not parse to Nautilus domain types.
104pub 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    /// Returns the base URL for this client.
135    #[must_use]
136    pub fn base_url(&self) -> &str {
137        &self.base_url
138    }
139
140    /// Returns a masked version of the API key for logging purposes.
141    #[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    /// Cancel all pending HTTP requests.
149    pub fn cancel_all_requests(&self) {
150        self.cancellation_token.read().cancel();
151    }
152
153    /// Replaces the cancelled token so new requests can proceed after reconnect.
154    pub fn reset_cancellation_token(&self) {
155        *self.cancellation_token.write() = CancellationToken::new();
156    }
157
158    /// Get a clone of the current cancellation token.
159    pub fn cancellation_token(&self) -> CancellationToken {
160        self.cancellation_token.read().clone()
161    }
162
163    /// Creates a new [`AxRawHttpClient`] using the default Ax HTTP URL.
164    ///
165    /// # Errors
166    ///
167    /// Returns an error if the retry manager cannot be created.
168    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    /// Creates a new [`AxRawHttpClient`] configured with credentials.
212    ///
213    /// # Errors
214    ///
215    /// Returns an error if the HTTP client cannot be created.
216    #[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    /// Sets the session token for authenticated requests.
263    ///
264    /// The session token is obtained through the login flow and used for bearer token authentication.
265    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        // Only retry idempotent methods to avoid duplicate orders/cancels
396        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    /// Fetches the current authenticated user information.
418    ///
419    /// # Endpoint
420    /// `GET /whoami`
421    ///
422    /// # Errors
423    ///
424    /// Returns an error if the request fails or the response cannot be parsed.
425    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    /// Fetches all available instruments.
431    ///
432    /// # Endpoint
433    /// `GET /instruments`
434    ///
435    /// # Errors
436    ///
437    /// Returns an error if the request fails or the response cannot be parsed.
438    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    /// Fetches all account balances for the authenticated user.
450    ///
451    /// # Endpoint
452    /// `GET /balances`
453    ///
454    /// # Errors
455    ///
456    /// Returns an error if the request fails or the response cannot be parsed.
457    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    /// Fetches all open positions for the authenticated user.
463    ///
464    /// # Endpoint
465    /// `GET /positions`
466    ///
467    /// # Errors
468    ///
469    /// Returns an error if the request fails or the response cannot be parsed.
470    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    /// Fetches all tickers.
476    ///
477    /// # Endpoint
478    /// `GET /tickers`
479    ///
480    /// # Errors
481    ///
482    /// Returns an error if the request fails or the response cannot be parsed.
483    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    /// Fetches tickers with optional pagination and sorting.
489    ///
490    /// # Endpoint
491    /// `GET /tickers`
492    ///
493    /// # Errors
494    ///
495    /// Returns an error if the request fails or the response cannot be parsed.
496    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    /// Fetches a single ticker by symbol.
505    ///
506    /// # Endpoint
507    /// `GET /ticker?symbol=<symbol>`
508    ///
509    /// # Errors
510    ///
511    /// Returns an error if the request fails or the response cannot be parsed.
512    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(&params), None, true)
515            .await
516            .map(|response| response.ticker)
517    }
518
519    /// Fetches a single instrument by symbol.
520    ///
521    /// # Endpoint
522    /// `GET /instrument?symbol=<symbol>`
523    ///
524    /// # Errors
525    ///
526    /// Returns an error if the request fails or the response cannot be parsed.
527    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(&params), None, false)
530            .await
531    }
532
533    /// Authenticates using API key and secret to obtain a session token.
534    ///
535    /// # Endpoint
536    /// `POST /authenticate`
537    ///
538    /// # Errors
539    ///
540    /// Returns an error if the request fails or the response cannot be parsed.
541    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    /// Authenticates using stored credentials or environment variables.
563    ///
564    /// # Credential Resolution
565    ///
566    /// Credentials are resolved in the following order:
567    /// 1. Stored credentials (from `with_credentials` constructor)
568    /// 2. Environment variables (`AX_API_KEY` and `AX_API_SECRET`)
569    ///
570    /// # Errors
571    ///
572    /// Returns an error if:
573    /// - No credentials are available from either source.
574    /// - The HTTP request fails.
575    /// - The credentials are invalid.
576    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    /// Places a new order.
608    ///
609    /// # Endpoint
610    /// `POST /place-order` (orders base URL)
611    ///
612    /// # Errors
613    ///
614    /// Returns an error if the request fails or the response cannot be parsed.
615    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    /// Cancels an existing order.
633    ///
634    /// # Endpoint
635    /// `POST /cancel-order` (orders base URL)
636    ///
637    /// # Errors
638    ///
639    /// Returns an error if the request fails or the response cannot be parsed.
640    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    /// Replaces (amends) an existing order.
656    ///
657    /// The exchange cancels the original order and creates a new one with the
658    /// updated fields. Unspecified optional fields inherit from the original.
659    ///
660    /// # Endpoint
661    /// `POST /replace-order` (orders base URL)
662    ///
663    /// # Errors
664    ///
665    /// Returns an error if the request fails or the response cannot be parsed.
666    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    /// Cancels all open orders, optionally filtered by account or symbol.
684    ///
685    /// # Endpoint
686    /// `POST /cancel-all-orders` (orders base URL)
687    ///
688    /// # Errors
689    ///
690    /// Returns an error if the request fails or the response cannot be parsed.
691    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    /// Fetches all open orders.
709    ///
710    /// # Endpoint
711    /// `GET /open-orders` (orders base URL)
712    ///
713    /// # Errors
714    ///
715    /// Returns an error if the request fails or the response cannot be parsed.
716    pub async fn get_open_orders(&self) -> Result<AxOpenOrdersResponse, AxHttpError> {
717        self.get_open_orders_page(&GetOpenOrdersParams::new()).await
718    }
719
720    /// Fetches one page of open orders.
721    ///
722    /// # Errors
723    ///
724    /// Returns an error if the request fails or the response cannot be parsed.
725    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    /// Fetches the default page of fills/trades.
741    ///
742    /// # Endpoint
743    /// `GET /fills`
744    ///
745    /// # Errors
746    ///
747    /// Returns an error if the request fails or the response cannot be parsed.
748    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(&params).await
755    }
756
757    /// Fetches one page of fills/trades.
758    ///
759    /// # Errors
760    ///
761    /// Returns an error if the request fails or the response cannot be parsed.
762    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    /// Fetches historical candles.
771    ///
772    /// # Endpoint
773    /// `GET /candles`
774    ///
775    /// # Errors
776    ///
777    /// Returns an error if the request fails or the response cannot be parsed.
778    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(&params),
791            None,
792            true,
793        )
794        .await
795    }
796
797    /// Fetches the current (incomplete) candle.
798    ///
799    /// # Endpoint
800    /// `GET /candles/current`
801    ///
802    /// # Errors
803    ///
804    /// Returns an error if the request fails or the response cannot be parsed.
805    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(&params),
816                None,
817                true,
818            )
819            .await?;
820        Ok(response.candle)
821    }
822
823    /// Fetches the last completed candle.
824    ///
825    /// # Endpoint
826    /// `GET /candles/last`
827    ///
828    /// # Errors
829    ///
830    /// Returns an error if the request fails or the response cannot be parsed.
831    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(&params),
842                None,
843                true,
844            )
845            .await?;
846        Ok(response.candle)
847    }
848
849    /// Fetches the default page of funding rates for a symbol.
850    ///
851    /// # Endpoint
852    /// `GET /funding-rates`
853    ///
854    /// # Errors
855    ///
856    /// Returns an error if the request fails or the response cannot be parsed.
857    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(&params).await
865    }
866
867    /// Fetches one page of funding rates for a symbol.
868    ///
869    /// # Errors
870    ///
871    /// Returns an error if the request fails or the response cannot be parsed.
872    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    /// Fetches the funding-slot schedule for a symbol on a trading day.
887    ///
888    /// # Endpoint
889    /// `GET /funding-slots`
890    ///
891    /// # Errors
892    ///
893    /// Returns an error if the request fails or the response cannot be parsed.
894    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    /// Fetches the current risk snapshot.
909    ///
910    /// # Endpoint
911    /// `GET /risk-snapshot`
912    ///
913    /// # Errors
914    ///
915    /// Returns an error if the request fails or the response cannot be parsed.
916    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    /// Previews an aggressive limit order to get the "take through" price.
928    ///
929    /// This endpoint calculates the price needed to sweep the order book for a given
930    /// quantity, which is used to simulate market orders on AX (which only supports
931    /// limit orders natively).
932    ///
933    /// # Endpoint
934    /// `POST /preview-aggressive-limit-order`
935    ///
936    /// # Errors
937    ///
938    /// Returns an error if the request fails or the response cannot be parsed.
939    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    /// Fetches the default page of transactions filtered by type.
956    ///
957    /// # Endpoint
958    /// `GET /transactions`
959    ///
960    /// # Errors
961    ///
962    /// Returns an error if the request fails or the response cannot be parsed.
963    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(&params).await
972    }
973
974    /// Fetches one page of transactions.
975    ///
976    /// # Errors
977    ///
978    /// Returns an error if the request fails or the response cannot be parsed.
979    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    /// Fetches recent trades for a symbol.
994    ///
995    /// # Endpoint
996    /// `GET /trades`
997    ///
998    /// # Errors
999    ///
1000    /// Returns an error if the request fails or the response cannot be parsed.
1001    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(&params), None, true)
1008            .await
1009    }
1010
1011    /// Fetches an order book snapshot for a symbol.
1012    ///
1013    /// # Endpoint
1014    /// `GET /book`
1015    ///
1016    /// # Errors
1017    ///
1018    /// Returns an error if the request fails or the response cannot be parsed.
1019    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        // The AX sandbox requires authentication for `/book` despite the public schema
1026        self.send_request::<AxBookResponse, _>(Method::GET, "/book", Some(&params), None, true)
1027            .await
1028    }
1029
1030    /// Fetches the status of a single order by order ID.
1031    ///
1032    /// # Endpoint
1033    /// `GET /order-status` (orders base URL)
1034    ///
1035    /// # Errors
1036    ///
1037    /// Returns an error if the request fails or the response cannot be parsed.
1038    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(&params),
1048            None,
1049            true,
1050        )
1051        .await
1052    }
1053
1054    /// Fetches the status of a single order by client order ID.
1055    ///
1056    /// # Endpoint
1057    /// `GET /order-status` (orders base URL)
1058    ///
1059    /// # Errors
1060    ///
1061    /// Returns an error if the request fails or the response cannot be parsed.
1062    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(&params),
1072            None,
1073            true,
1074        )
1075        .await
1076    }
1077
1078    /// Fetches historical orders with optional filters.
1079    ///
1080    /// # Endpoint
1081    /// `GET /orders` (orders base URL)
1082    ///
1083    /// # Errors
1084    ///
1085    /// Returns an error if the request fails or the response cannot be parsed.
1086    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    /// Checks the initial margin requirement for a proposed order.
1102    ///
1103    /// # Endpoint
1104    /// `POST /initial-margin-requirement` (orders base URL)
1105    ///
1106    /// # Errors
1107    ///
1108    /// Returns an error if the request fails or the response cannot be parsed.
1109    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/// High-level HTTP client for the Ax REST API.
1128///
1129/// This client wraps the underlying [`AxRawHttpClient`] to provide a convenient
1130/// interface for Python bindings and instrument caching.
1131#[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    /// Creates a new [`AxHttpClient`] using the default Ax HTTP URL.
1169    ///
1170    /// # Errors
1171    ///
1172    /// Returns an error if the retry manager cannot be created.
1173    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    /// Creates a new [`AxHttpClient`] configured with credentials.
1200    ///
1201    /// # Errors
1202    ///
1203    /// Returns an error if the HTTP client cannot be created.
1204    #[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    /// Returns the base URL for this client.
1236    #[must_use]
1237    pub fn base_url(&self) -> &str {
1238        self.inner.base_url()
1239    }
1240
1241    /// Returns a masked version of the API key for logging purposes.
1242    #[must_use]
1243    pub fn api_key_masked(&self) -> String {
1244        self.inner.api_key_masked()
1245    }
1246
1247    /// Cancel all pending HTTP requests.
1248    pub fn cancel_all_requests(&self) {
1249        self.inner.cancel_all_requests();
1250    }
1251
1252    /// Replaces the cancelled token so new requests can proceed after reconnect.
1253    pub fn reset_cancellation_token(&self) {
1254        self.inner.reset_cancellation_token();
1255    }
1256
1257    /// Sets the session token for authenticated requests.
1258    ///
1259    /// The session token is obtained through the login flow and used for bearer token authentication.
1260    pub fn set_session_token(&self, token: SecretString) {
1261        self.inner.set_session_token(token);
1262    }
1263
1264    /// Generates a timestamp for initialization.
1265    fn generate_ts_init(&self) -> UnixNanos {
1266        self.clock.get_time_ns()
1267    }
1268
1269    /// Checks if the client is initialized.
1270    ///
1271    /// The client is considered initialized if any instruments have been cached from the venue.
1272    #[must_use]
1273    pub fn is_initialized(&self) -> bool {
1274        self.cache_initialized.load(Ordering::Acquire)
1275    }
1276
1277    /// Returns a snapshot of all instrument symbols currently held in the internal cache.
1278    #[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    /// Caches multiple instruments.
1288    ///
1289    /// Any existing instruments with the same symbols will be replaced.
1290    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    /// Caches a single instrument.
1300    ///
1301    /// Any existing instrument with the same symbol will be replaced.
1302    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    /// Authenticates with Ax using API credentials.
1309    ///
1310    /// On success, the session token is automatically stored for subsequent authenticated requests.
1311    ///
1312    /// # Errors
1313    ///
1314    /// Returns an error if the HTTP request fails or credentials are invalid.
1315    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    /// Authenticates using stored credentials or environment variables.
1331    ///
1332    /// # Credential Resolution
1333    ///
1334    /// Credentials are resolved in the following order:
1335    /// 1. Stored credentials (from `with_credentials` constructor)
1336    /// 2. Environment variables (`AX_API_KEY` and `AX_API_SECRET`)
1337    ///
1338    /// On success, the session token is automatically stored for subsequent authenticated requests.
1339    ///
1340    /// # Errors
1341    ///
1342    /// Returns an error if:
1343    /// - No credentials are available from either source.
1344    /// - The HTTP request fails.
1345    /// - The credentials are invalid.
1346    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    /// Gets an instrument from the cache by symbol.
1357    pub fn get_instrument(&self, symbol: &Ustr) -> Option<InstrumentAny> {
1358        self.instruments_cache.get_cloned(symbol)
1359    }
1360
1361    /// Resolves the maker and taker fee rates for the account behind the current credentials.
1362    ///
1363    /// AX reports fee rates per account rather than per user, and returns the accounts the
1364    /// credentials can act on. The first entry's rates are returned and serve as the default
1365    /// for fill payloads without an account identifier. Each successful call replaces the
1366    /// cached rates for every listed account that supplies both maker and taker rates.
1367    /// Instruments do not carry these rates.
1368    ///
1369    /// Requires an authenticated client.
1370    ///
1371    /// # Errors
1372    ///
1373    /// Returns an error if the request fails, the response carries no accounts, or the selected
1374    /// account supplies no fee rates. An absent rate is not treated as zero, because a zero rate
1375    /// is itself valid and a silent zero would outlive the response that caused it.
1376    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    /// Requests all instruments from Ax.
1416    ///
1417    /// # Errors
1418    ///
1419    /// Returns an error if the HTTP request fails or instrument parsing fails.
1420    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            // Skip test instruments (not real tradable products)
1437            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    /// Requests a single instrument from Ax by symbol.
1454    ///
1455    /// # Errors
1456    ///
1457    /// Returns an error if the HTTP request fails or instrument parsing fails.
1458    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    /// Requests an order book snapshot from Ax and builds a Nautilus [`OrderBook`].
1471    ///
1472    /// Requires the instrument to be cached.
1473    ///
1474    /// # Errors
1475    ///
1476    /// Returns an error if:
1477    /// - The instrument is not found in the cache.
1478    /// - The HTTP request fails.
1479    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    /// Requests recent trades from Ax and parses them to Nautilus [`TradeTick`].
1536    ///
1537    /// The AX trades endpoint does not accept time range parameters, so
1538    /// `start` and `end` are applied as client-side filters after fetching.
1539    ///
1540    /// Requires the instrument to be cached.
1541    ///
1542    /// # Errors
1543    ///
1544    /// Returns an error if:
1545    /// - The instrument is not found in the cache.
1546    /// - The HTTP request fails.
1547    /// - Trade parsing fails.
1548    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    /// Requests historical bars from Ax and parses them to Nautilus Bar types.
1590    ///
1591    /// Requires the instrument to be cached (call `request_instruments` first).
1592    ///
1593    /// # Errors
1594    ///
1595    /// Returns an error if:
1596    /// - The instrument is not found in the cache.
1597    /// - The HTTP request fails.
1598    /// - Bar parsing fails.
1599    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    /// Requests funding rates from Ax and parses them to Nautilus types.
1638    ///
1639    /// Traverses the provider's cursor chain. This is a best-effort historical
1640    /// read, not an atomic snapshot if AX corrects rows during the traversal.
1641    ///
1642    /// # Errors
1643    ///
1644    /// Returns an error if the HTTP request fails.
1645    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(&params).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    /// Requests the funding-slot schedule for a symbol on a trading day.
1793    ///
1794    /// AX returns a single response covering the whole trading day, so there
1795    /// is no pagination. The schedule has no Nautilus domain equivalent, so
1796    /// the venue response is returned verbatim.
1797    ///
1798    /// # Errors
1799    ///
1800    /// Returns an error if the HTTP request fails or the response cannot be parsed.
1801    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(&params).await
1814    }
1815
1816    /// Requests account state from Ax and parses to a Nautilus [`AccountState`].
1817    ///
1818    /// # Errors
1819    ///
1820    /// Returns an error if the HTTP request fails or parsing fails.
1821    pub async fn request_account_state(
1822        &self,
1823        account_id: AccountId,
1824    ) -> anyhow::Result<AccountState> {
1825        // Time-bound the snapshot so a hung /risk-snapshot cannot stall connect
1826        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    /// Checks the initial margin requirement for a proposed order.
1860    ///
1861    /// # Errors
1862    ///
1863    /// Returns an error if the HTTP request fails.
1864    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    /// Queries a single order by venue order ID or client order ID using the
1877    /// dedicated `/order-status` endpoint, which works for any order state.
1878    ///
1879    /// The caller must supply `order_side`, `order_type`, and `time_in_force`
1880    /// because the endpoint does not return these fields.
1881    /// Canceled, expired, and rejected orders with no remaining quantity use `/orders`
1882    /// to recover their original quantity.
1883    ///
1884    /// # Errors
1885    ///
1886    /// Returns an error if:
1887    /// - Neither `venue_order_id` nor `client_order_id` is provided.
1888    /// - The HTTP request fails.
1889    /// - The original quantity is unavailable in order history.
1890    #[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    /// Requests open orders from Ax and parses them to Nautilus [`OrderStatusReport`].
1984    ///
1985    /// Missing instruments are requested from Ax and cached before parsing order details.
1986    ///
1987    /// The `cid_resolver` parameter is an optional function that resolves a `cid` (u64)
1988    /// to a `ClientOrderId`. This is needed for correlating orders submitted via WebSocket.
1989    ///
1990    /// # Errors
1991    ///
1992    /// Returns an error if:
1993    /// - The HTTP request fails.
1994    /// - An order's instrument cannot be fetched or parsed.
1995    /// - An order cannot be mapped to a complete status report.
1996    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(&params)
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    /// Requests historical orders from Ax and parses them to Nautilus
2120    /// [`OrderStatusReport`].
2121    ///
2122    /// Missing instruments are requested from Ax and cached before parsing order details.
2123    ///
2124    /// The `cid_resolver` parameter is an optional function that resolves a `cid` (u64)
2125    /// to a `ClientOrderId`. This is needed for correlating orders submitted via WebSocket.
2126    ///
2127    /// # Errors
2128    ///
2129    /// Returns an error if:
2130    /// - The HTTP request or pagination contract fails.
2131    /// - An order's instrument cannot be fetched or parsed.
2132    /// - An order cannot be mapped to a complete status report.
2133    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(&params)
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    /// Requests fills from Ax and parses them to Nautilus [`FillReport`].
2204    ///
2205    /// Missing instruments are requested from Ax and cached before parsing fill details.
2206    /// Traverses the provider's cursor chain. This is a best-effort historical
2207    /// read, not an atomic snapshot if AX corrects rows during the traversal.
2208    ///
2209    /// # Errors
2210    ///
2211    /// Returns an error if:
2212    /// - The HTTP request fails.
2213    /// - A fill's instrument cannot be fetched or parsed.
2214    /// - Fill parsing fails.
2215    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        // The AX `/fills` endpoint requires a bounded time range and caps the span at 7 days
2224        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(&params)
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    /// Requests positions from Ax and parses them to Nautilus [`PositionStatusReport`].
2336    ///
2337    /// Missing instruments are requested from Ax and cached before parsing position details.
2338    ///
2339    /// # Errors
2340    ///
2341    /// Returns an error if:
2342    /// - The HTTP request fails.
2343    /// - A position's instrument cannot be fetched or parsed.
2344    ///
2345    /// # Notes
2346    ///
2347    /// Position parsing failures are skipped with a warning.
2348    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            // Skip flat positions (zero quantity)
2363            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    /// Cancels all open orders for an instrument.
2393    ///
2394    /// # Errors
2395    ///
2396    /// Returns an error if the request fails.
2397    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}