nautilus_derive/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//! REST client for the Derive API using the shared HTTP transport.
17//!
18//! [`DeriveHttpClient`] exposes typed `send_public` / `send_private`
19//! dispatchers plus thin wrappers for the two endpoints that establish the
20//! plumbing this crate needs to grow against: `public/get_all_instruments` and
21//! `private/order`. Authenticated requests inject the EIP-191 session-key
22//! headers built by [`crate::signing::auth`].
23
24use std::{
25    collections::HashSet,
26    fmt::Debug,
27    sync::{
28        Arc,
29        atomic::{AtomicU64, Ordering},
30    },
31};
32
33use ahash::AHashMap;
34use alloy::signers::local::PrivateKeySigner;
35use nautilus_core::string::secret::{REDACTED, SecretString, zeroize_json_value};
36use nautilus_network::{
37    http::{
38        HttpClient, HttpClientError, HttpRedirectPolicy, HttpResponse, Method, RETRY_AFTER,
39        create_standard_nautilus_headers, parse_retry_after,
40    },
41    retry::{RetryConfig, RetryManager},
42};
43use serde::{Serialize, de::DeserializeOwned};
44use serde_json::Value;
45use ustr::Ustr;
46
47use crate::{
48    common::{
49        consts::{
50            HEADER_DERIVE_SIGNATURE, HEADER_DERIVE_TIMESTAMP, HEADER_DERIVE_WALLET,
51            HTTP_MAX_RETRIES, HTTP_RETRY_DELAY_INITIAL_MS, HTTP_RETRY_DELAY_MAX_MS, HTTP_TIMEOUT,
52        },
53        enums::DeriveInstrumentType,
54        rate_limit::{self, DeriveRateLimiter},
55        retry::{http_retry_config, should_retry_http_error},
56    },
57    http::{
58        error::{DeriveHttpError, Result},
59        models::{
60            DeriveCancelByLabelResult, DeriveEmptyResult, DeriveInstrument,
61            DeriveInstrumentsResult, DeriveOpenOrdersResult, DeriveOrder, DeriveOrderResult,
62            DeriveOrdersResult, DerivePositionsResult, DerivePublicCandle,
63            DerivePublicFundingRateHistoryResult, DerivePublicTradesResult, DeriveReplaceOutcome,
64            DeriveReplaceResult, DeriveSubaccount, DeriveTickerSnapshot, DeriveTickersResult,
65            DeriveTradesResult, JsonRpcResponse,
66        },
67        query::{
68            DeriveCancelAllParams, DeriveCancelByLabelParams, DeriveCancelParams,
69            DeriveGetOpenOrdersParams, DeriveGetOrderHistoryParams, DeriveGetOrderParams,
70            DeriveGetPositionsParams, DeriveGetSubaccountParams, DeriveGetTradeHistoryParams,
71            DeriveGetTriggerOrdersParams, DeriveOrderParams, DeriveReplaceParams, PaginationCursor,
72        },
73    },
74    signing::auth::{AuthHeaders, build_rest_auth_headers},
75};
76
77/// Credentials used to sign authenticated REST requests.
78///
79/// `Debug` is implemented manually so the session key never escapes through
80/// loggers or Python `__repr__`.
81#[derive(Clone)]
82pub struct DeriveCredentials {
83    /// Owner's EOA or multisig address (`0x`-prefixed hex, 42 chars).
84    pub wallet_address: String,
85    /// secp256k1 session-key signer.
86    pub signer: PrivateKeySigner,
87}
88
89impl DeriveCredentials {
90    /// Constructs credentials by parsing `session_key_hex` into a signer.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error if the session-key hex cannot be parsed.
95    pub fn new(wallet_address: impl Into<String>, session_key_hex: &str) -> Result<Self> {
96        let signer: PrivateKeySigner = session_key_hex
97            .parse()
98            .map_err(|e| DeriveHttpError::decode(format!("invalid session key: {e}")))?;
99        Ok(Self {
100            wallet_address: wallet_address.into(),
101            signer,
102        })
103    }
104}
105
106impl Debug for DeriveCredentials {
107    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108        f.debug_struct(stringify!(DeriveCredentials))
109            .field("wallet_address", &self.wallet_address)
110            .field("signer", &REDACTED)
111            .finish()
112    }
113}
114
115/// HTTP client for the Derive REST API.
116///
117/// The client carries an atomic `id` counter so every request frame has a
118/// unique correlator; the REST transport ships only `params` on the wire but
119/// the id is preserved for logs and reused by the upcoming WebSocket client.
120/// Each call routes through a [`RetryManager`] that re-signs auth headers on
121/// every attempt, so retries never replay a stale `X-DeriveTimestamp`.
122#[derive(Debug, Clone)]
123pub struct DeriveHttpClient {
124    client: HttpClient,
125    base_url: String,
126    credentials: Option<DeriveCredentials>,
127    next_id: Arc<AtomicU64>,
128    timeout_secs: u64,
129    retry_manager: Arc<RetryManager<DeriveHttpError>>,
130    rate_limiter: Arc<DeriveRateLimiter>,
131}
132
133impl DeriveHttpClient {
134    /// Creates a public-only client.
135    ///
136    /// When `retry_config` is `None`, uses [`HTTP_MAX_RETRIES`],
137    /// [`HTTP_RETRY_DELAY_INITIAL_MS`], and [`HTTP_RETRY_DELAY_MAX_MS`].
138    ///
139    /// # Errors
140    ///
141    /// Returns an error if the underlying HTTP client (proxy URL, TLS init) cannot be constructed.
142    pub fn new(
143        base_url: impl Into<String>,
144        timeout_secs: Option<u64>,
145        proxy_url: Option<String>,
146        retry_config: Option<RetryConfig>,
147    ) -> Result<Self> {
148        let timeout_secs = timeout_secs.unwrap_or_else(|| HTTP_TIMEOUT.as_secs());
149        let client = build_client(timeout_secs, proxy_url)?;
150        let base_url = trim_trailing_slash(base_url.into());
151        let rate_limiter = Arc::new(DeriveRateLimiter::shared(&base_url, None, None, None));
152
153        let retry_config = retry_config.unwrap_or_else(|| {
154            http_retry_config(
155                HTTP_MAX_RETRIES,
156                HTTP_RETRY_DELAY_INITIAL_MS,
157                HTTP_RETRY_DELAY_MAX_MS,
158            )
159        });
160
161        Ok(Self {
162            client,
163            base_url,
164            credentials: None,
165            next_id: Arc::new(AtomicU64::new(1)),
166            timeout_secs,
167            retry_manager: Arc::new(RetryManager::new(retry_config)),
168            rate_limiter,
169        })
170    }
171
172    /// Creates a client with credentials installed for `send_private` calls.
173    ///
174    /// # Errors
175    ///
176    /// Returns an error if the underlying HTTP client cannot be constructed.
177    pub fn with_credentials(
178        base_url: impl Into<String>,
179        credentials: DeriveCredentials,
180        timeout_secs: Option<u64>,
181        proxy_url: Option<String>,
182        retry_config: Option<RetryConfig>,
183    ) -> Result<Self> {
184        let mut client = Self::new(base_url, timeout_secs, proxy_url, retry_config)?;
185
186        client.rate_limiter = Arc::new(DeriveRateLimiter::shared(
187            &client.base_url,
188            Some(&credentials.wallet_address),
189            None,
190            None,
191        ));
192        client.credentials = Some(credentials);
193        Ok(client)
194    }
195
196    /// Returns the configured base URL (no trailing slash).
197    #[must_use]
198    pub fn base_url(&self) -> &str {
199        &self.base_url
200    }
201
202    /// Returns `true` when credentials are installed.
203    #[must_use]
204    pub fn has_credentials(&self) -> bool {
205        self.credentials.is_some()
206    }
207
208    /// Allocates the next correlator id.
209    fn next_id(&self) -> u64 {
210        self.next_id.fetch_add(1, Ordering::Relaxed)
211    }
212
213    /// Sends an unauthenticated request and decodes the JSON-RPC envelope.
214    ///
215    /// Public endpoints are idempotent reads; this path retries transient
216    /// failures via the configured [`RetryManager`].
217    ///
218    /// # Errors
219    ///
220    /// Returns an error if transport, HTTP, or JSON-RPC processing fails.
221    pub async fn send_public<P, R>(&self, method: &str, params: &P) -> Result<R>
222    where
223        P: Serialize + ?Sized,
224        R: DeserializeOwned,
225    {
226        let id = self.next_id();
227        self.dispatch(method, params, id, false, true, None).await
228    }
229
230    /// Sends an authenticated idempotent request (private reads).
231    ///
232    /// Used for `private/get_*` endpoints whose responses are pure reads of
233    /// venue state. Transient failures retry via the configured
234    /// [`RetryManager`].
235    ///
236    /// # Errors
237    ///
238    /// Returns an error if the client was built without credentials. Other variants propagate from
239    /// the transport or the venue.
240    pub async fn send_private<P, R>(&self, method: &str, params: &P) -> Result<R>
241    where
242        P: Serialize + ?Sized,
243        R: DeserializeOwned,
244    {
245        if self.credentials.is_none() {
246            return Err(DeriveHttpError::MissingCredentials {
247                method: method.to_owned(),
248            });
249        }
250
251        let id = self.next_id();
252        self.dispatch(method, params, id, true, true, None).await
253    }
254
255    /// Sends an authenticated request exactly once (no retry).
256    ///
257    /// Used for state-changing endpoints (`private/order`, `private/cancel`,
258    /// `private/cancel_all`, `private/cancel_by_label`, `private/replace`)
259    /// where a transport-level failure leaves the venue's view of the
260    /// signed action ambiguous: the request may have been accepted before
261    /// the network broke. Automatic replay would either double-submit (when
262    /// the venue accepted) or trigger a duplicate-nonce rejection (which
263    /// the caller would surface as `OrderRejected` even though the original
264    /// is live). Callers are expected to resolve ambiguous outcomes via
265    /// reconciliation rather than retry here.
266    ///
267    /// Instrument pacing uses the typed wrapper's instrument or the body's
268    /// `instrument_name` field. Requests without an instrument consume only
269    /// their wallet request budgets.
270    ///
271    /// # Errors
272    ///
273    /// Returns an error if the client was built without credentials. Other variants propagate from
274    /// the transport or the venue.
275    pub async fn send_private_once<P, R>(&self, method: &str, params: &P) -> Result<R>
276    where
277        P: Serialize + ?Sized,
278        R: DeserializeOwned,
279    {
280        if self.credentials.is_none() {
281            return Err(DeriveHttpError::MissingCredentials {
282                method: method.to_owned(),
283            });
284        }
285
286        let id = self.next_id();
287        self.dispatch(method, params, id, true, false, None).await
288    }
289
290    /// Sends an authenticated matching-engine write exactly once, pacing it
291    /// against both the wallet and per-instrument allowances.
292    async fn send_private_write<P, R>(
293        &self,
294        method: &str,
295        params: &P,
296        instrument_name: Ustr,
297    ) -> Result<R>
298    where
299        P: Serialize + ?Sized,
300        R: DeserializeOwned,
301    {
302        if self.credentials.is_none() {
303            return Err(DeriveHttpError::MissingCredentials {
304                method: method.to_owned(),
305            });
306        }
307
308        let id = self.next_id();
309        self.dispatch(method, params, id, true, false, Some(instrument_name))
310            .await
311    }
312
313    /// Fetches the venue's listed instruments.
314    ///
315    /// Fetches every page for `currency` and `instrument_type`. For options,
316    /// `expired` selects either live (`false`) or expired (`true`) listings.
317    /// Perpetuals and spot ignore `expired`.
318    ///
319    /// # Errors
320    ///
321    /// Returns an error if transport, HTTP, or JSON-RPC processing fails. Returns a decode error
322    /// for invalid pagination counts or a page count that exceeds the collection limit.
323    pub async fn get_instruments(
324        &self,
325        currency: &str,
326        instrument_type: DeriveInstrumentType,
327        expired: bool,
328    ) -> Result<Vec<DeriveInstrument>> {
329        let mut instruments = Vec::new();
330        let mut names = HashSet::new();
331        let mut pages = PaginationCursor::new_public();
332
333        loop {
334            let params = serde_json::json!({
335                "currency": currency,
336                "instrument_type": instrument_type,
337                "expired": expired,
338                "page": pages.page(),
339                "page_size": 1000,
340            });
341            let result: DeriveInstrumentsResult = self
342                .send_public("public/get_all_instruments", &params)
343                .await?;
344            let more = pages.advance(
345                &result.pagination,
346                result
347                    .instruments
348                    .iter()
349                    .map(|row| row.instrument_name.as_str()),
350            )?;
351            instruments.extend(
352                result
353                    .instruments
354                    .into_iter()
355                    .filter(|row| names.insert(row.instrument_name)),
356            );
357
358            if !more {
359                return Ok(instruments);
360            }
361        }
362    }
363
364    /// Fetches a single instrument definition by name.
365    ///
366    /// Expired options remain queryable by name. Settlement prices appear
367    /// here before they appear in the bulk listing.
368    ///
369    /// # Errors
370    ///
371    /// Returns an error if transport, HTTP, or JSON-RPC processing fails. Returns a decode error
372    /// when the response names a different instrument.
373    pub async fn get_instrument(&self, instrument_name: &str) -> Result<DeriveInstrument> {
374        let params = serde_json::json!({
375            "instrument_name": instrument_name,
376        });
377        let instrument: DeriveInstrument =
378            self.send_public("public/get_instrument", &params).await?;
379
380        if instrument.instrument_name.as_str() != instrument_name {
381            return Err(DeriveHttpError::decode(format!(
382                "instrument response identity mismatch: requested {instrument_name}, received {}",
383                instrument.instrument_name,
384            )));
385        }
386
387        Ok(instrument)
388    }
389
390    /// Fetches a page of public trade history for the instrument.
391    ///
392    /// `from_timestamp` / `to_timestamp` are UNIX milliseconds and bound the
393    /// returned window. `page` is 1-indexed; `page_size` is capped by the venue
394    /// at 1000.
395    ///
396    /// # Errors
397    ///
398    /// Returns an error if transport, HTTP, or JSON-RPC processing fails.
399    pub async fn get_trade_history(
400        &self,
401        instrument_name: &str,
402        from_timestamp: Option<i64>,
403        to_timestamp: Option<i64>,
404        page: u32,
405        page_size: u32,
406    ) -> Result<DerivePublicTradesResult> {
407        let mut params = serde_json::Map::new();
408        params.insert("instrument_name".to_string(), instrument_name.into());
409        params.insert("page".to_string(), page.into());
410        params.insert("page_size".to_string(), page_size.into());
411        if let Some(from) = from_timestamp {
412            params.insert("from_timestamp".to_string(), from.into());
413        }
414
415        if let Some(to) = to_timestamp {
416            params.insert("to_timestamp".to_string(), to.into());
417        }
418
419        self.send_public("public/get_trade_history", &Value::Object(params))
420            .await
421    }
422
423    /// Fetches the public funding rate history for the instrument.
424    ///
425    /// `start_timestamp` / `end_timestamp` are UNIX milliseconds. `period`, if
426    /// provided, selects the sample interval in seconds.
427    ///
428    /// # Errors
429    ///
430    /// Returns an error if transport, HTTP, or JSON-RPC processing fails.
431    pub async fn get_funding_rate_history(
432        &self,
433        instrument_name: &str,
434        start_timestamp: Option<i64>,
435        end_timestamp: Option<i64>,
436        period: Option<u32>,
437    ) -> Result<DerivePublicFundingRateHistoryResult> {
438        let mut params = serde_json::Map::new();
439        params.insert("instrument_name".to_string(), instrument_name.into());
440        if let Some(start) = start_timestamp {
441            params.insert("start_timestamp".to_string(), start.into());
442        }
443
444        if let Some(end) = end_timestamp {
445            params.insert("end_timestamp".to_string(), end.into());
446        }
447
448        if let Some(period) = period {
449            params.insert("period".to_string(), period.into());
450        }
451
452        self.send_public("public/get_funding_rate_history", &Value::Object(params))
453            .await
454    }
455
456    /// Fetches OHLCV candles via `public/get_tradingview_chart_data`.
457    ///
458    /// `start_timestamp` / `end_timestamp` are UNIX **seconds** and bound the
459    /// returned window. `period` is the bucket size in seconds; the venue
460    /// accepts 60, 300, 900, 1800, 3600, 14400, 28800, 86400, and 604800.
461    /// The venue ships `result` as a flat array; the client decodes it
462    /// directly into `Vec<DerivePublicCandle>`.
463    ///
464    /// # Errors
465    ///
466    /// Returns an error if transport, HTTP, or JSON-RPC processing fails.
467    pub async fn get_candles(
468        &self,
469        instrument_name: &str,
470        start_timestamp: i64,
471        end_timestamp: i64,
472        period: u32,
473    ) -> Result<Vec<DerivePublicCandle>> {
474        let params = serde_json::json!({
475            "instrument_name": instrument_name,
476            "start_timestamp": start_timestamp,
477            "end_timestamp": end_timestamp,
478            "period": period,
479        });
480        self.send_public("public/get_tradingview_chart_data", &params)
481            .await
482    }
483
484    /// Fetches current ticker snapshots.
485    ///
486    /// `currency` is the underlying (`"ETH"`, `"BTC"`, etc.). Options require
487    /// both `currency` and `expiry_date`; perps and ERC-20 spot pairs reject
488    /// `expiry_date`.
489    ///
490    /// # Errors
491    ///
492    /// Returns an error if transport, HTTP, or JSON-RPC processing fails.
493    pub async fn get_tickers(
494        &self,
495        instrument_type: DeriveInstrumentType,
496        currency: Option<&str>,
497        expiry_date: Option<&str>,
498    ) -> Result<DeriveTickersResult> {
499        let mut params = serde_json::Map::new();
500        params.insert(
501            "instrument_type".to_string(),
502            serde_json::to_value(instrument_type).map_err(DeriveHttpError::from)?,
503        );
504
505        if let Some(currency) = currency {
506            params.insert("currency".to_string(), currency.into());
507        }
508
509        if let Some(expiry_date) = expiry_date {
510            params.insert("expiry_date".to_string(), expiry_date.into());
511        }
512
513        self.send_public("public/get_tickers", &Value::Object(params))
514            .await
515    }
516
517    /// Fetches the current ticker snapshot for one instrument.
518    ///
519    /// This is a single-instrument convenience wrapper over
520    /// `public/get_tickers`, which replaced Derive's deprecated
521    /// `public/get_ticker` RPC.
522    ///
523    /// # Errors
524    ///
525    /// Returns an error if transport, HTTP, or JSON-RPC processing fails, or the response omits the
526    /// requested instrument.
527    pub async fn get_ticker(&self, instrument_name: &str) -> Result<DeriveTickerSnapshot> {
528        let request = ticker_request(instrument_name)?;
529        let result = self
530            .get_tickers(
531                request.instrument_type,
532                Some(request.currency),
533                request.expiry_date,
534            )
535            .await?;
536
537        let mut ticker = result
538            .tickers
539            .get(instrument_name)
540            .cloned()
541            .ok_or_else(|| {
542                DeriveHttpError::decode(format!(
543                    "missing ticker `{instrument_name}` in public/get_tickers response"
544                ))
545            })?;
546
547        ticker.instrument_name = instrument_name.into();
548        Ok(ticker)
549    }
550
551    /// Submits a signed order to the venue.
552    ///
553    /// `params` must be the fully-built signed `private/order` body.
554    ///
555    /// # Errors
556    ///
557    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
558    /// errors.
559    pub async fn submit_order(&self, params: &DeriveOrderParams) -> Result<DeriveOrder> {
560        let result: DeriveOrderResult = self
561            .send_private_write("private/order", params, params.instrument_name)
562            .await?;
563        Ok(result.order)
564    }
565
566    /// Cancels a single order by venue order id.
567    ///
568    /// # Errors
569    ///
570    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
571    /// errors.
572    pub async fn cancel_order(&self, params: &DeriveCancelParams) -> Result<DeriveEmptyResult> {
573        self.send_private_write("private/cancel", params, params.instrument_name)
574            .await
575    }
576
577    /// Cancels every open order on the subaccount, optionally scoped to an
578    /// instrument.
579    ///
580    /// # Errors
581    ///
582    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
583    /// errors.
584    pub async fn cancel_all(&self, params: &DeriveCancelAllParams) -> Result<DeriveEmptyResult> {
585        self.send_private_once("private/cancel_all", params).await
586    }
587
588    /// Cancels every open order for the given user label on the subaccount.
589    ///
590    /// # Errors
591    ///
592    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
593    /// errors.
594    pub async fn cancel_by_label(
595        &self,
596        params: &DeriveCancelByLabelParams,
597    ) -> Result<DeriveCancelByLabelResult> {
598        self.send_private_once("private/cancel_by_label", params)
599            .await
600    }
601
602    /// Submits a signed `private/replace` request that cancels one order before
603    /// creating its replacement.
604    ///
605    /// `params` must be the fully-built typed request body.
606    ///
607    /// # Errors
608    ///
609    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
610    /// errors.
611    pub async fn replace_order(
612        &self,
613        params: &DeriveReplaceParams,
614    ) -> Result<DeriveReplaceOutcome> {
615        let result: DeriveReplaceResult = self
616            .send_private_write("private/replace", params, params.order.instrument_name)
617            .await?;
618        result
619            .into_outcome(
620                &params.order_id_to_cancel,
621                &params.order.label,
622                params.order.envelope.subaccount_id,
623                params.order.instrument_name.as_str(),
624            )
625            .map_err(DeriveHttpError::decode)
626    }
627
628    /// Returns the subaccount snapshot including margin, balances, and
629    /// open orders.
630    ///
631    /// # Errors
632    ///
633    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
634    /// errors. Returns a decode error if a response subaccount ID differs from the request.
635    pub async fn get_subaccount(
636        &self,
637        params: &DeriveGetSubaccountParams,
638    ) -> Result<DeriveSubaccount> {
639        let result: DeriveSubaccount = self.send_private("private/get_subaccount", params).await?;
640        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
641        validate_order_subaccounts(params.subaccount_id, &result.open_orders)?;
642        Ok(result)
643    }
644
645    /// Returns currently open orders for the subaccount.
646    ///
647    /// # Errors
648    ///
649    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
650    /// errors. Returns a decode error if a response subaccount ID differs from the request.
651    pub async fn get_open_orders(
652        &self,
653        params: &DeriveGetOpenOrdersParams,
654    ) -> Result<DeriveOpenOrdersResult> {
655        let result: DeriveOpenOrdersResult =
656            self.send_private("private/get_open_orders", params).await?;
657        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
658        validate_order_subaccounts(params.subaccount_id, &result.orders)?;
659        Ok(result)
660    }
661
662    /// Returns currently untriggered trigger orders for the subaccount.
663    ///
664    /// # Errors
665    ///
666    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
667    /// errors. Returns a decode error if a response subaccount ID differs from the request.
668    pub async fn get_trigger_orders(
669        &self,
670        params: &DeriveGetTriggerOrdersParams,
671    ) -> Result<DeriveOpenOrdersResult> {
672        let result: DeriveOpenOrdersResult = self
673            .send_private("private/get_trigger_orders", params)
674            .await?;
675        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
676        validate_order_subaccounts(params.subaccount_id, &result.orders)?;
677        Ok(result)
678    }
679
680    /// Returns a single order by venue order id.
681    ///
682    /// # Errors
683    ///
684    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
685    /// errors. Returns a decode error if a response subaccount ID differs from the request.
686    pub async fn get_order(&self, params: &DeriveGetOrderParams) -> Result<DeriveOrder> {
687        let result: DeriveOrder = self.send_private("private/get_order", params).await?;
688        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
689        Ok(result)
690    }
691
692    /// Returns one page of order history for the subaccount, optionally
693    /// scoped to an instrument and time window.
694    ///
695    /// `from_timestamp` / `to_timestamp` are UNIX milliseconds. `page` is
696    /// 1-indexed and `page_size` is capped by the venue at 1000.
697    ///
698    /// # Errors
699    ///
700    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
701    /// errors. Returns a decode error if a response subaccount ID differs from the request.
702    pub async fn get_order_history(
703        &self,
704        params: &DeriveGetOrderHistoryParams,
705    ) -> Result<DeriveOrdersResult> {
706        let result: DeriveOrdersResult = self
707            .send_private("private/get_order_history", params)
708            .await?;
709        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
710        validate_order_subaccounts(params.subaccount_id, &result.orders)?;
711        Ok(result)
712    }
713
714    /// Returns one page of subaccount trade history.
715    ///
716    /// # Errors
717    ///
718    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
719    /// errors. Returns a decode error if a response subaccount ID differs from the request.
720    pub async fn get_private_trade_history(
721        &self,
722        params: &DeriveGetTradeHistoryParams,
723    ) -> Result<DeriveTradesResult> {
724        let result: DeriveTradesResult = self
725            .send_private("private/get_trade_history", params)
726            .await?;
727        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
728        for trade in &result.trades {
729            validate_subaccount_id(params.subaccount_id, trade.subaccount_id)?;
730        }
731
732        Ok(result)
733    }
734
735    /// Returns the positions held by the subaccount.
736    ///
737    /// # Errors
738    ///
739    /// Returns an error if no credentials were installed; otherwise propagates transport and venue
740    /// errors. Returns a decode error if a response subaccount ID differs from the request.
741    pub async fn get_positions(
742        &self,
743        params: &DeriveGetPositionsParams,
744    ) -> Result<DerivePositionsResult> {
745        let result: DerivePositionsResult =
746            self.send_private("private/get_positions", params).await?;
747        validate_subaccount_id(params.subaccount_id, result.subaccount_id)?;
748        Ok(result)
749    }
750
751    async fn dispatch<P, R>(
752        &self,
753        method: &str,
754        params: &P,
755        id: u64,
756        authenticate: bool,
757        retry: bool,
758        instrument_name: Option<Ustr>,
759    ) -> Result<R>
760    where
761        P: Serialize + ?Sized,
762        R: DeserializeOwned,
763    {
764        let url = format!("{}/{}", self.base_url, method.trim_start_matches('/'));
765        let mut body_value = serde_json::to_value(params).map_err(DeriveHttpError::from)?;
766        let body = serde_json::to_string(&body_value).map_err(DeriveHttpError::from);
767
768        let rate_class = rate_limit::rate_class_for_method(method);
769
770        let instrument_name = instrument_name.or_else(|| {
771            body_value
772                .get("instrument_name")
773                .and_then(Value::as_str)
774                .map(Ustr::from)
775        });
776
777        zeroize_json_value(&mut body_value);
778        let body = SecretString::from(body?);
779
780        // Sign per-attempt so the venue never sees a stale `X-DeriveTimestamp`
781        // after a long backoff window; single-shot writes still run the
782        // closure once and use freshly built headers. The pacing wait
783        // happens inside the closure, so pacing delays never consume the
784        // signed timestamp's validity.
785        let attempt = || async {
786            if authenticate {
787                self.rate_limiter
788                    .await_class_ready(rate_class, instrument_name.as_ref())
789                    .await;
790            } else {
791                self.rate_limiter.await_public_ready().await;
792            }
793
794            let mut headers: AHashMap<String, String> = AHashMap::with_capacity(4);
795            headers.insert("Content-Type".to_string(), "application/json".to_string());
796
797            if authenticate {
798                let auth = self.build_auth_headers(method)?;
799                headers.insert(HEADER_DERIVE_WALLET.to_string(), auth.wallet);
800                headers.insert(HEADER_DERIVE_TIMESTAMP.to_string(), auth.timestamp);
801                headers.insert(
802                    HEADER_DERIVE_SIGNATURE.to_string(),
803                    auth.signature.into_inner(),
804                );
805            }
806
807            let response = self
808                .client
809                .request_with_secret_body(
810                    Method::POST,
811                    url.clone(),
812                    None,
813                    Some(headers.into_iter().collect()),
814                    body.clone(),
815                    Some(self.timeout_secs),
816                    None,
817                )
818                .await
819                .map_err(DeriveHttpError::from)?;
820
821            decode_envelope(method, id, response)
822        };
823
824        if retry {
825            self.retry_manager
826                .invocation(
827                    method,
828                    attempt,
829                    should_retry_http_error,
830                    DeriveHttpError::Retry,
831                )
832                .retry_delay(&DeriveHttpError::retry_after)
833                .execute()
834                .await
835        } else {
836            attempt().await
837        }
838    }
839
840    fn build_auth_headers(&self, method: &str) -> Result<AuthHeaders> {
841        let credentials =
842            self.credentials
843                .as_ref()
844                .ok_or_else(|| DeriveHttpError::MissingCredentials {
845                    method: method.to_owned(),
846                })?;
847
848        let auth = build_rest_auth_headers(&credentials.wallet_address, &credentials.signer)?;
849        Ok(auth)
850    }
851}
852
853#[derive(Debug, Clone, Copy)]
854struct TickerRequest<'a> {
855    instrument_type: DeriveInstrumentType,
856    currency: &'a str,
857    expiry_date: Option<&'a str>,
858}
859
860fn ticker_request(instrument_name: &str) -> Result<TickerRequest<'_>> {
861    let Some((currency, suffix)) = instrument_name.split_once('-') else {
862        return Err(DeriveHttpError::decode(format!(
863            "invalid Derive instrument name `{instrument_name}`"
864        )));
865    };
866
867    if suffix == "PERP" {
868        return Ok(TickerRequest {
869            instrument_type: DeriveInstrumentType::Perp,
870            currency,
871            expiry_date: None,
872        });
873    }
874
875    let mut parts = suffix.split('-');
876
877    let Some(expiry_date) = parts.next() else {
878        return Ok(TickerRequest {
879            instrument_type: DeriveInstrumentType::Erc20,
880            currency,
881            expiry_date: None,
882        });
883    };
884
885    let has_option_tail = parts.clone().count() == 2;
886    if expiry_date.len() == 8 && expiry_date.chars().all(|c| c.is_ascii_digit()) && has_option_tail
887    {
888        return Ok(TickerRequest {
889            instrument_type: DeriveInstrumentType::Option,
890            currency,
891            expiry_date: Some(expiry_date),
892        });
893    }
894
895    Ok(TickerRequest {
896        instrument_type: DeriveInstrumentType::Erc20,
897        currency,
898        expiry_date: None,
899    })
900}
901
902fn build_client(
903    timeout_secs: u64,
904    proxy_url: Option<String>,
905) -> std::result::Result<HttpClient, HttpClientError> {
906    HttpClient::builder()
907        .redirect_policy(HttpRedirectPolicy::Reject)
908        .headers(create_standard_nautilus_headers().into_iter().collect())
909        .header_keys(vec![RETRY_AFTER.as_str().to_string()])
910        .timeout_secs(timeout_secs)
911        .maybe_proxy_url(proxy_url)
912        .rate_limiters(Vec::new())
913        .build()
914}
915
916fn trim_trailing_slash(url: String) -> String {
917    if url.ends_with('/') {
918        url.trim_end_matches('/').to_string()
919    } else {
920        url
921    }
922}
923
924fn decode_envelope<R: DeserializeOwned>(
925    method: &str,
926    request_id: u64,
927    response: HttpResponse,
928) -> Result<R> {
929    let status = response.status.as_u16();
930
931    let retry_after = response
932        .headers
933        .get(RETRY_AFTER.as_str())
934        .and_then(|value| {
935            parse_retry_after(
936                value,
937                nautilus_core::UnixNanos::from(nautilus_core::time::nanos_since_unix_epoch())
938                    .to_datetime_utc(),
939            )
940        });
941
942    let is_success_status = (200..300).contains(&status);
943    let body = response.body;
944
945    let envelope: JsonRpcResponse<R> = match serde_json::from_slice(&body) {
946        Ok(env) => env,
947        Err(e) => {
948            if !is_success_status {
949                let text = String::from_utf8_lossy(&body).into_owned();
950                return Err(DeriveHttpError::Http {
951                    status,
952                    message: truncate(text, 512),
953                    retry_after,
954                });
955            }
956
957            return Err(DeriveHttpError::decode(format!(
958                "failed to decode `{method}` response: {e}",
959            )));
960        }
961    };
962
963    if let Some(err) = envelope.error {
964        return Err(DeriveHttpError::JsonRpc {
965            code: err.code,
966            message: err.message,
967            data: err.data,
968            retry_after,
969        });
970    }
971
972    // Gateways (Cloudflare, the wallet auth proxy) return non-2xx with a JSON body
973    // like {"message": "Unauthorized"} that parses into an empty envelope. Surface
974    // those as Http errors so retry/reconcile logic sees the real status code
975    // instead of MissingResult.
976    if !is_success_status {
977        let text = String::from_utf8_lossy(&body).into_owned();
978        return Err(DeriveHttpError::Http {
979            status,
980            message: truncate(text, 512),
981            retry_after,
982        });
983    }
984
985    if let Some(echoed) = envelope.id
986        && echoed != request_id
987    {
988        log::debug!(
989            "derive: id mismatch for `{method}` (sent={request_id}, recv={echoed}); accepting result",
990        );
991    }
992
993    envelope
994        .result
995        .ok_or_else(|| DeriveHttpError::MissingResult {
996            method: method.to_owned(),
997        })
998}
999
1000fn truncate(s: String, max: usize) -> String {
1001    if s.len() <= max {
1002        return s;
1003    }
1004
1005    let mut cutoff = max;
1006    while cutoff > 0 && !s.is_char_boundary(cutoff) {
1007        cutoff -= 1;
1008    }
1009
1010    let mut out = String::with_capacity(cutoff + 3);
1011    out.push_str(&s[..cutoff]);
1012    out.push_str("...");
1013    out
1014}
1015
1016fn validate_order_subaccounts(expected: u64, orders: &[DeriveOrder]) -> Result<()> {
1017    for order in orders {
1018        validate_subaccount_id(expected, order.subaccount_id)?;
1019    }
1020
1021    Ok(())
1022}
1023
1024fn validate_subaccount_id(expected: u64, received: i64) -> Result<()> {
1025    if u64::try_from(received).ok() != Some(expected) {
1026        return Err(DeriveHttpError::decode(format!(
1027            "subaccount response identity mismatch: requested {expected}, received {received}",
1028        )));
1029    }
1030
1031    Ok(())
1032}
1033
1034#[cfg(test)]
1035mod tests {
1036    use std::collections::HashMap;
1037
1038    use nautilus_network::http::{HttpStatus, StatusCode};
1039    use nautilus_testkit::http::assert_http_redirect_rejected;
1040    use rstest::rstest;
1041
1042    use super::*;
1043
1044    const SESSION_KEY_HEX: &str =
1045        "0x2ae8be44db8a590d20bffbe3b6872df9b569147d3bf6801a35a28281a4816bbd";
1046    const TEST_WALLET: &str = "0x000000000000000000000000000000000000aaaa";
1047
1048    fn test_client() -> DeriveHttpClient {
1049        DeriveHttpClient::new("https://api.example/", None, None, None).expect("client builds")
1050    }
1051
1052    fn test_response(status: u16, body: &serde_json::Value) -> HttpResponse {
1053        let status_code = StatusCode::from_u16(status).unwrap();
1054
1055        HttpResponse {
1056            status: HttpStatus::new(status_code),
1057            headers: HashMap::new(),
1058            body: serde_json::to_vec(body).unwrap().into(),
1059        }
1060    }
1061
1062    #[tokio::test]
1063    async fn test_authenticated_client_rejects_redirects() {
1064        let client = build_client(3, None).unwrap();
1065        assert_http_redirect_rejected(|url| async move {
1066            client
1067                .get(url, None, None, Some(3), None)
1068                .await
1069                .unwrap()
1070                .status
1071                .as_u16()
1072        })
1073        .await;
1074    }
1075
1076    #[rstest]
1077    fn test_credentials_debug_redacts_signer() {
1078        let creds = DeriveCredentials::new(TEST_WALLET, SESSION_KEY_HEX).unwrap();
1079        let dbg = format!("{creds:?}");
1080        assert!(dbg.contains(REDACTED));
1081        assert!(dbg.contains(TEST_WALLET));
1082        assert!(!dbg.contains(SESSION_KEY_HEX));
1083    }
1084
1085    #[rstest]
1086    fn test_credentials_rejects_invalid_session_key() {
1087        let err = DeriveCredentials::new(TEST_WALLET, "not-hex").expect_err("must reject");
1088        match err {
1089            DeriveHttpError::Decode(msg) => assert!(msg.contains("invalid session key")),
1090            other => panic!("expected Decode, was {other:?}"),
1091        }
1092    }
1093
1094    #[rstest]
1095    fn test_base_url_trims_trailing_slash() {
1096        let client = test_client();
1097        assert_eq!(client.base_url(), "https://api.example");
1098    }
1099
1100    #[rstest]
1101    fn test_new_has_no_credentials() {
1102        assert!(!test_client().has_credentials());
1103    }
1104
1105    #[rstest]
1106    fn test_with_credentials_sets_creds() {
1107        let creds = DeriveCredentials::new(TEST_WALLET, SESSION_KEY_HEX).unwrap();
1108        let client =
1109            DeriveHttpClient::with_credentials("https://api.example", creds, None, None, None)
1110                .unwrap();
1111        assert!(client.has_credentials());
1112    }
1113
1114    #[rstest]
1115    fn test_next_id_increments_monotonically() {
1116        let client = test_client();
1117        let a = client.next_id();
1118        let b = client.next_id();
1119        let c = client.next_id();
1120        assert_eq!(b, a + 1);
1121        assert_eq!(c, b + 1);
1122    }
1123
1124    #[rstest]
1125    fn test_decode_envelope_returns_result() {
1126        let resp = test_response(200, &serde_json::json!({"id": 1, "result": {"ok": true}}));
1127        let value: Value = decode_envelope("public/get_instruments", 1, resp).unwrap();
1128        assert_eq!(value["ok"], true);
1129    }
1130
1131    #[rstest]
1132    fn test_decode_envelope_accepts_null_empty_result() {
1133        let resp = test_response(200, &serde_json::json!({"id": 1, "result": null}));
1134        let result: DeriveEmptyResult = decode_envelope("private/cancel", 1, resp).unwrap();
1135        assert_eq!(result, DeriveEmptyResult {});
1136    }
1137
1138    #[rstest]
1139    fn test_decode_envelope_propagates_jsonrpc_error() {
1140        let resp = test_response(
1141            200,
1142            &serde_json::json!({
1143                "id": 1,
1144                "error": {"code": -32601, "message": "Method not found"}
1145            }),
1146        );
1147        let err: DeriveHttpError = decode_envelope::<Value>("public/missing", 1, resp).unwrap_err();
1148        match err {
1149            DeriveHttpError::JsonRpc { code, message, .. } => {
1150                assert_eq!(code, -32601);
1151                assert_eq!(message, "Method not found");
1152            }
1153            other => panic!("expected JsonRpc, was {other:?}"),
1154        }
1155    }
1156
1157    #[rstest]
1158    fn test_decode_envelope_flags_missing_result() {
1159        let resp = test_response(200, &serde_json::json!({"id": 1}));
1160        let err = decode_envelope::<Value>("public/get_instruments", 1, resp).unwrap_err();
1161        assert!(matches!(err, DeriveHttpError::MissingResult { .. }));
1162    }
1163
1164    #[rstest]
1165    fn test_decode_envelope_flags_non_2xx_with_unparsable_body() {
1166        let status_code = StatusCode::from_u16(503).unwrap();
1167
1168        let response = HttpResponse {
1169            status: HttpStatus::new(status_code),
1170            headers: HashMap::new(),
1171            body: bytes::Bytes::from_static(b"<html>upstream down</html>"),
1172        };
1173
1174        let err = decode_envelope::<Value>("public/get_instruments", 1, response).unwrap_err();
1175        match err {
1176            DeriveHttpError::Http {
1177                status, message, ..
1178            } => {
1179                assert_eq!(status, 503);
1180                assert!(message.contains("upstream down"));
1181            }
1182            other => panic!("expected Http, was {other:?}"),
1183        }
1184    }
1185
1186    #[rstest]
1187    fn test_decode_envelope_flags_non_2xx_with_non_envelope_json() {
1188        // Gateways return non-2xx with JSON bodies like {"message": "Unauthorized"}.
1189        // These parse as an empty JsonRpcResponse; the status must still surface.
1190        let resp = test_response(401, &serde_json::json!({"message": "Unauthorized"}));
1191        let err = decode_envelope::<Value>("private/order", 1, resp).unwrap_err();
1192        match err {
1193            DeriveHttpError::Http {
1194                status, message, ..
1195            } => {
1196                assert_eq!(status, 401);
1197                assert!(message.contains("Unauthorized"));
1198            }
1199            other => panic!("expected Http, was {other:?}"),
1200        }
1201    }
1202
1203    #[rstest]
1204    fn test_decode_envelope_prefers_jsonrpc_error_over_http_status() {
1205        // When the venue returns a proper JSON-RPC error envelope with a non-2xx
1206        // status, the envelope wins because it carries richer venue context.
1207        let status_code = StatusCode::from_u16(400).unwrap();
1208        let body = serde_json::json!({
1209            "id": 1,
1210            "error": {"code": -32602, "message": "Invalid params"},
1211        });
1212
1213        let response = HttpResponse {
1214            status: HttpStatus::new(status_code),
1215            headers: HashMap::new(),
1216            body: serde_json::to_vec(&body).unwrap().into(),
1217        };
1218
1219        let err = decode_envelope::<Value>("private/order", 1, response).unwrap_err();
1220        assert!(matches!(err, DeriveHttpError::JsonRpc { code: -32602, .. }));
1221    }
1222
1223    #[rstest]
1224    fn test_truncate_handles_multi_byte_char_at_boundary() {
1225        // "Ω" is two bytes (0xCE 0xA9). Truncating to a length that lands mid-glyph
1226        // must not panic; we step back to the prior char boundary.
1227        let s = "ΩΩΩΩΩΩΩΩΩΩ".to_string();
1228        assert_eq!(s.len(), 20);
1229        let out = truncate(s, 5);
1230        assert!(out.ends_with("..."));
1231        let prefix = out.trim_end_matches("...");
1232        assert!(prefix.is_char_boundary(prefix.len()));
1233        assert!(prefix.chars().all(|c| c == 'Ω'));
1234    }
1235
1236    #[rstest]
1237    fn test_truncate_returns_input_when_under_limit() {
1238        let s = "short".to_string();
1239        assert_eq!(truncate(s, 16), "short");
1240    }
1241
1242    #[rstest]
1243    fn test_decode_envelope_non_2xx_body_with_non_ascii_does_not_panic() {
1244        // Regression: a Cloudflare-style 503 page containing non-ASCII bytes near
1245        // the truncation cutoff must not panic.
1246        let glyph = "Ω";
1247        let body = glyph.repeat(600);
1248        let status_code = StatusCode::from_u16(503).unwrap();
1249
1250        let response = HttpResponse {
1251            status: HttpStatus::new(status_code),
1252            headers: HashMap::new(),
1253            body: body.into_bytes().into(),
1254        };
1255
1256        let err = decode_envelope::<Value>("public/get_instruments", 1, response).unwrap_err();
1257        assert!(matches!(err, DeriveHttpError::Http { status: 503, .. }));
1258    }
1259
1260    #[rstest]
1261    fn test_decode_envelope_accepts_id_mismatch() {
1262        let resp = test_response(200, &serde_json::json!({"id": 99, "result": "ok"}));
1263        let value: Value = decode_envelope("public/get_instruments", 1, resp).unwrap();
1264        assert_eq!(value, serde_json::json!("ok"));
1265    }
1266
1267    #[tokio::test]
1268    async fn test_send_private_without_credentials_errors() {
1269        let client = test_client();
1270        let err = client
1271            .send_private::<_, Value>("private/order", &serde_json::json!({}))
1272            .await
1273            .expect_err("must require credentials");
1274
1275        match err {
1276            DeriveHttpError::MissingCredentials { method } => {
1277                assert_eq!(method, "private/order");
1278            }
1279            other => panic!("expected MissingCredentials, was {other:?}"),
1280        }
1281    }
1282}
nautilus_derive/http/client.rs

nautilus_derive/http/
client.rs
