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

nautilus_derive/http/
client.rs
