nautilus_derive/websocket/
messages.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//! Wire payloads for the Derive WebSocket JSON-RPC transport.
17//!
18//! The transport reuses the [`crate::http::models::JsonRpcRequest`] /
19//! [`crate::http::models::JsonRpcResponse`] envelope; this module covers only
20//! the params payloads and the inbound notification frame.
21
22use std::{
23    collections::HashMap,
24    fmt::{Debug, Display},
25    str::FromStr,
26};
27
28#[cfg(test)]
29use nautilus_core::string::secret::REDACTED;
30use nautilus_core::{serialization::deserialize_decimal_token, string::secret::SecretString};
31use nautilus_model::identifiers::InstrumentId;
32use rust_decimal::Decimal;
33use serde::{Deserialize, Serialize};
34use serde_json::{Value, value::RawValue};
35use ustr::Ustr;
36use zeroize::Zeroize;
37
38use crate::{
39    common::{
40        enums::{
41            DeriveInstrumentType, DeriveOrderbookDepth, DeriveOrderbookGroup, DeriveTickerInterval,
42        },
43        parse::{format_instrument_id, salvage_rows},
44    },
45    http::models::{
46        DeriveAggregateTradingStats, DeriveOptionPricing, DeriveOrder, DerivePublicTrade,
47        DeriveTicker, DeriveTickerSnapshot, DeriveTrade, JsonRpcError,
48    },
49};
50
51pub(crate) const DEFAULT_ORDERBOOK_GROUP: &str = "1";
52pub(crate) const DEFAULT_ORDERBOOK_DEPTH: &str = "10";
53pub(crate) const DEFAULT_TICKER_INTERVAL: &str = "1000";
54
55/// Params payload for `public/login`.
56///
57/// The wallet/timestamp/signature triple comes from
58/// [`crate::signing::auth::build_ws_login`]; the signature recovers the owner
59/// or a registered, unexpired session key for `wallet` over the millisecond timestamp string.
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Zeroize)]
61pub struct WsLoginParams {
62    /// Owner's EOA or multisig address (`0x`-prefixed hex).
63    pub wallet: String,
64    /// Millisecond UNIX timestamp string (matches the bytes that were signed).
65    pub timestamp: String,
66    /// 0x-prefixed signature hex over `timestamp` under EIP-191.
67    pub signature: SecretString,
68}
69
70/// Params payload for `subscribe`.
71#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
72pub struct WsSubscribeParams {
73    /// Channel topics to subscribe to (e.g. `ticker_slim.ETH-PERP.1000`).
74    pub channels: Vec<DeriveWsChannel>,
75}
76
77/// Params payload for `unsubscribe`.
78#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
79pub struct WsUnsubscribeParams {
80    /// Channel topics to drop.
81    pub channels: Vec<DeriveWsChannel>,
82}
83
84/// Params for the persisted, wallet-wide `private/set_cancel_on_disconnect` setting.
85#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
86pub struct WsSetCancelOnDisconnectParams {
87    /// Owner EOA or multisig address of the authenticated session.
88    pub wallet: String,
89    /// Whether the venue enables cancel-on-disconnect for the wallet.
90    pub enabled: bool,
91}
92
93/// Derive WebSocket subscription channel topic.
94#[derive(Debug, Clone, PartialEq, Eq, Hash)]
95pub enum DeriveWsChannel {
96    /// Public compact ticker channel.
97    TickerSlim {
98        /// Venue instrument name.
99        instrument_name: Ustr,
100        /// Update interval in milliseconds.
101        interval: DeriveTickerInterval,
102    },
103    /// Public order book channel.
104    Orderbook {
105        /// Venue instrument name.
106        instrument_name: Ustr,
107        /// Venue grouping increment.
108        group: DeriveOrderbookGroup,
109        /// Requested depth.
110        depth: DeriveOrderbookDepth,
111    },
112    /// Public trades channel.
113    Trades {
114        /// Venue instrument type.
115        instrument_type: DeriveInstrumentType,
116        /// Venue currency.
117        currency: Ustr,
118    },
119    /// Private order updates channel.
120    Orders {
121        /// Subaccount id.
122        subaccount_id: u64,
123    },
124    /// Private trade updates channel.
125    PrivateTrades {
126        /// Subaccount id.
127        subaccount_id: u64,
128    },
129    /// Private balance updates channel.
130    Balances {
131        /// Subaccount id.
132        subaccount_id: u64,
133    },
134    /// Passthrough topic for venue channels not yet modeled by the adapter.
135    Raw(String),
136}
137
138impl DeriveWsChannel {
139    /// Returns a compact ticker channel.
140    #[must_use]
141    pub fn ticker_slim(instrument_name: impl AsRef<str>, interval: impl AsRef<str>) -> Self {
142        let instrument_name = instrument_name.as_ref();
143        let interval = interval.as_ref();
144
145        let Ok(interval) = DeriveTickerInterval::from_str(interval) else {
146            return Self::Raw(ticker_channel(instrument_name, interval));
147        };
148
149        Self::TickerSlim {
150            instrument_name: Ustr::from(instrument_name),
151            interval,
152        }
153    }
154
155    /// Returns an order book channel.
156    #[must_use]
157    pub fn orderbook(
158        instrument_name: impl AsRef<str>,
159        group: impl AsRef<str>,
160        depth: impl AsRef<str>,
161    ) -> Self {
162        let instrument_name = instrument_name.as_ref();
163        let group = group.as_ref();
164        let depth = depth.as_ref();
165
166        let Ok(group) = DeriveOrderbookGroup::from_str(group) else {
167            return Self::Raw(orderbook_channel(instrument_name, group, depth));
168        };
169
170        let Ok(depth) = DeriveOrderbookDepth::from_str(depth) else {
171            return Self::Raw(orderbook_channel(instrument_name, group.as_ref(), depth));
172        };
173
174        Self::Orderbook {
175            instrument_name: Ustr::from(instrument_name),
176            group,
177            depth,
178        }
179    }
180
181    /// Returns a public trades channel.
182    #[must_use]
183    pub fn trades(instrument_type: impl AsRef<str>, currency: impl AsRef<str>) -> Self {
184        let instrument_type = instrument_type.as_ref();
185        let currency = currency.as_ref();
186
187        let Ok(instrument_type) = DeriveInstrumentType::from_str(instrument_type) else {
188            return Self::Raw(trades_channel(instrument_type, currency));
189        };
190
191        Self::Trades {
192            instrument_type,
193            currency: Ustr::from(currency),
194        }
195    }
196
197    /// Returns a private orders channel.
198    #[must_use]
199    pub const fn orders(subaccount_id: u64) -> Self {
200        Self::Orders { subaccount_id }
201    }
202
203    /// Returns a private trades channel.
204    #[must_use]
205    pub const fn private_trades(subaccount_id: u64) -> Self {
206        Self::PrivateTrades { subaccount_id }
207    }
208
209    /// Returns a private balances channel.
210    #[must_use]
211    pub const fn balances(subaccount_id: u64) -> Self {
212        Self::Balances { subaccount_id }
213    }
214
215    /// Parses a topic string into the known channel family when possible.
216    #[must_use]
217    pub fn from_topic(topic: impl Into<String>) -> Self {
218        let topic = topic.into();
219
220        if let Some(rest) = topic.strip_prefix("ticker_slim.")
221            && let Some((instrument_name, interval)) = rest.rsplit_once('.')
222            && !instrument_name.is_empty()
223            && !interval.is_empty()
224        {
225            return Self::ticker_slim(instrument_name, interval);
226        }
227
228        if let Some(rest) = topic.strip_prefix("orderbook.")
229            && let Some((rest, depth)) = rest.rsplit_once('.')
230            && let Some((instrument_name, group)) = rest.rsplit_once('.')
231            && !instrument_name.is_empty()
232            && !group.is_empty()
233            && !depth.is_empty()
234        {
235            return Self::orderbook(instrument_name, group, depth);
236        }
237
238        if let Some(rest) = topic.strip_prefix("trades.")
239            && let Some((instrument_type, currency)) = rest.split_once('.')
240            && !instrument_type.is_empty()
241            && !currency.is_empty()
242        {
243            return Self::trades(instrument_type, currency);
244        }
245
246        if let Some((subaccount_id, suffix)) = topic.split_once('.')
247            && let Ok(subaccount_id) = subaccount_id.parse::<u64>()
248        {
249            return match suffix {
250                "orders" => Self::orders(subaccount_id),
251                "trades" => Self::private_trades(subaccount_id),
252                "balances" => Self::balances(subaccount_id),
253                _ => Self::Raw(topic),
254            };
255        }
256
257        Self::Raw(topic)
258    }
259}
260
261impl Display for DeriveWsChannel {
262    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
263        match self {
264            Self::TickerSlim {
265                instrument_name,
266                interval,
267            } => f.write_str(&ticker_channel(instrument_name.as_str(), interval.as_ref())),
268            Self::Orderbook {
269                instrument_name,
270                group,
271                depth,
272            } => f.write_str(&orderbook_channel(
273                instrument_name.as_str(),
274                group.as_ref(),
275                depth.as_ref(),
276            )),
277            Self::Trades {
278                instrument_type,
279                currency,
280            } => f.write_str(&trades_channel(instrument_type.as_ref(), currency.as_str())),
281            Self::Orders { subaccount_id } => f.write_str(&orders_channel(*subaccount_id)),
282            Self::PrivateTrades { subaccount_id } => {
283                f.write_str(&private_trades_channel(*subaccount_id))
284            }
285            Self::Balances { subaccount_id } => f.write_str(&balances_channel(*subaccount_id)),
286            Self::Raw(topic) => f.write_str(topic),
287        }
288    }
289}
290
291impl Serialize for DeriveWsChannel {
292    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
293    where
294        S: serde::Serializer,
295    {
296        serializer.serialize_str(&self.to_string())
297    }
298}
299
300impl<'de> Deserialize<'de> for DeriveWsChannel {
301    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
302    where
303        D: serde::Deserializer<'de>,
304    {
305        String::deserialize(deserializer).map(Self::from_topic)
306    }
307}
308
309impl From<String> for DeriveWsChannel {
310    fn from(value: String) -> Self {
311        Self::from_topic(value)
312    }
313}
314
315impl From<&str> for DeriveWsChannel {
316    fn from(value: &str) -> Self {
317        Self::from_topic(value)
318    }
319}
320
321/// Method-specific params accepted by the WebSocket JSON-RPC request path.
322#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
323#[serde(untagged)]
324pub enum WsRequestParams {
325    /// Params for `public/login`.
326    Login(WsLoginParams),
327    /// Params for `subscribe`.
328    Subscribe(WsSubscribeParams),
329    /// Params for `unsubscribe`.
330    Unsubscribe(WsUnsubscribeParams),
331}
332
333impl From<WsLoginParams> for WsRequestParams {
334    fn from(value: WsLoginParams) -> Self {
335        Self::Login(value)
336    }
337}
338
339impl From<WsSubscribeParams> for WsRequestParams {
340    fn from(value: WsSubscribeParams) -> Self {
341        Self::Subscribe(value)
342    }
343}
344
345impl From<WsUnsubscribeParams> for WsRequestParams {
346    fn from(value: WsUnsubscribeParams) -> Self {
347        Self::Unsubscribe(value)
348    }
349}
350
351/// Authorized subaccount IDs returned by `public/login`.
352pub type WsLoginResult = Vec<u64>;
353
354#[derive(Debug)]
355pub(super) struct WsSetCancelOnDisconnectResult;
356
357impl<'de> Deserialize<'de> for WsSetCancelOnDisconnectResult {
358    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
359    where
360        D: serde::Deserializer<'de>,
361    {
362        let value = String::deserialize(deserializer)?;
363        if value != "ok" {
364            return Err(serde::de::Error::custom(
365                "expected cancel-on-disconnect result \"ok\"",
366            ));
367        }
368
369        Ok(Self)
370    }
371}
372
373/// Result payload returned by `subscribe`.
374#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)]
375pub struct WsSubscribeResult {
376    /// Current subscriptions reported by the venue.
377    #[serde(rename = "current_subscriptions")]
378    pub channels: Vec<DeriveWsChannel>,
379    /// Per-channel subscription status reported by the venue.
380    pub status: HashMap<DeriveWsChannel, Ustr>,
381}
382
383/// Result payload returned by `unsubscribe`.
384#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)]
385pub struct WsUnsubscribeResult {
386    /// Channels that remain subscribed after the operation.
387    pub remaining_subscriptions: Vec<DeriveWsChannel>,
388    /// Per-channel unsubscribe status reported by the venue.
389    pub status: HashMap<DeriveWsChannel, Ustr>,
390}
391
392/// Inbound notification frame pushed by the venue on a subscribed channel.
393///
394/// The venue tags the frame with `method = "subscription"` and inlines the
395/// channel-specific payload under `params.data`.
396#[derive(Debug, Clone, Deserialize)]
397pub struct WsSubscriptionFrame {
398    /// Routing key (`method` on the wire). Always `"subscription"`.
399    #[serde(default)]
400    pub method: Option<Ustr>,
401    /// Subscription envelope.
402    pub params: WsSubscriptionPayload,
403}
404
405/// Channel-tagged notification payload nested under [`WsSubscriptionFrame::params`].
406///
407/// The channel payload is held as a [`RawValue`] (the raw JSON bytes) rather
408/// than a decoded [`Value`]; each channel parser decodes those bytes straight
409/// into its typed struct, so the inbound path never materializes the payload
410/// into an intermediate `Value` tree.
411#[derive(Debug, Clone, Deserialize)]
412pub struct WsSubscriptionPayload {
413    /// Channel that produced the update (e.g. `"ticker_slim.ETH-PERP.1000"`).
414    pub channel: Ustr,
415    /// Opaque per-channel payload; specific channels decode this further.
416    pub data: Box<RawValue>,
417}
418
419/// Price level in a Derive order book snapshot.
420///
421/// The venue sends levels as `[price, amount]` tuples with decimal strings.
422#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
423pub struct DeriveOrderbookLevel(
424    /// Price level.
425    #[serde(deserialize_with = "deserialize_decimal_token")]
426    pub Decimal,
427    /// Aggregated amount at the price level.
428    #[serde(deserialize_with = "deserialize_decimal_token")]
429    pub Decimal,
430);
431
432impl DeriveOrderbookLevel {
433    /// Returns the level price.
434    #[must_use]
435    pub const fn price(&self) -> Decimal {
436        self.0
437    }
438
439    /// Returns the level amount.
440    #[must_use]
441    pub const fn amount(&self) -> Decimal {
442        self.1
443    }
444}
445
446/// Order book snapshot pushed on `orderbook.{instrument_name}.{group}.{depth}`.
447#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
448pub struct DeriveOrderbookData {
449    /// Instrument name on the Derive venue.
450    pub instrument_name: Ustr,
451    /// Snapshot timestamp in UNIX milliseconds.
452    pub timestamp: i64,
453    /// Bid price levels, best first.
454    pub bids: Vec<DeriveOrderbookLevel>,
455    /// Ask price levels, best first.
456    pub asks: Vec<DeriveOrderbookLevel>,
457}
458
459impl DeriveOrderbookData {
460    /// Returns the Nautilus instrument ID for this Derive symbol.
461    ///
462    /// # Errors
463    ///
464    /// Returns an error if the venue symbol is empty or contains only whitespace.
465    pub fn instrument_id(&self) -> anyhow::Result<InstrumentId> {
466        format_instrument_id(self.instrument_name)
467    }
468}
469
470/// Channel-tagged order book update.
471#[derive(Debug, Clone, PartialEq, Eq)]
472pub struct DeriveOrderbookMsg {
473    /// Channel that produced the update.
474    pub channel: Ustr,
475    /// Parsed order book data.
476    pub data: DeriveOrderbookData,
477}
478
479/// Channel-tagged public trades update.
480#[derive(Debug, Clone)]
481pub struct DeriveTradesMsg {
482    /// Channel that produced the update.
483    pub channel: Ustr,
484    /// Trades carried by the update.
485    pub trades: Vec<DerivePublicTrade>,
486}
487
488/// Private `{subaccount_id}.orders` subscription payload.
489#[derive(Debug, Clone)]
490pub struct DeriveOrdersSubscriptionData {
491    /// Orders carried by the update.
492    pub orders: Vec<DeriveOrder>,
493}
494
495impl<'de> Deserialize<'de> for DeriveOrdersSubscriptionData {
496    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
497    where
498        D: serde::Deserializer<'de>,
499    {
500        Ok(Self {
501            orders: subscription_rows(deserializer)?,
502        })
503    }
504}
505
506/// Private `{subaccount_id}.trades` subscription payload.
507#[derive(Debug, Clone)]
508pub struct DeriveTradesSubscriptionData {
509    /// Trades carried by the update.
510    pub trades: Vec<DeriveTrade>,
511}
512
513impl<'de> Deserialize<'de> for DeriveTradesSubscriptionData {
514    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
515    where
516        D: serde::Deserializer<'de>,
517    {
518        Ok(Self {
519            trades: subscription_rows(deserializer)?,
520        })
521    }
522}
523
524/// Decodes a private subscription payload that arrives as either a single row
525/// object or an array of rows; array elements are salvaged per element.
526fn subscription_rows<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
527where
528    D: serde::Deserializer<'de>,
529    T: serde::de::DeserializeOwned,
530{
531    let raw = Box::<RawValue>::deserialize(deserializer)?;
532    if raw.get().starts_with('[') {
533        let rows = serde_json::from_str(raw.get()).map_err(serde::de::Error::custom)?;
534        Ok(salvage_rows(rows))
535    } else {
536        serde_json::from_str(raw.get())
537            .map(|row| vec![row])
538            .map_err(serde::de::Error::custom)
539    }
540}
541
542/// Ticker payload shape pushed by the Derive ticker channels.
543#[derive(Debug, Clone)]
544pub enum DeriveTickerData {
545    /// Full ticker shape with a feed timestamp and nested ticker snapshot.
546    Envelope {
547        /// Feed snapshot timestamp in UNIX milliseconds.
548        timestamp: i64,
549        /// Full instrument ticker snapshot.
550        instrument_ticker: DeriveTicker,
551    },
552    /// Compact ticker shape with a feed timestamp and nested ticker snapshot.
553    SlimEnvelope {
554        /// Feed snapshot timestamp in UNIX milliseconds.
555        timestamp: i64,
556        /// Compact instrument ticker snapshot.
557        instrument_ticker: DeriveTickerSnapshot,
558    },
559    /// Legacy shape where `params.data` is the ticker object itself.
560    Ticker(DeriveTicker),
561}
562
563impl<'de> Deserialize<'de> for DeriveTickerData {
564    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
565        #[derive(Deserialize)]
566        struct Envelope<T> {
567            timestamp: i64,
568            instrument_ticker: T,
569        }
570        let raw = Box::<RawValue>::deserialize(deserializer)?;
571        if let Ok(envelope) = serde_json::from_str::<Envelope<DeriveTicker>>(raw.get()) {
572            return Ok(Self::Envelope {
573                timestamp: envelope.timestamp,
574                instrument_ticker: envelope.instrument_ticker,
575            });
576        }
577
578        if let Ok(envelope) = serde_json::from_str::<Envelope<DeriveTickerSnapshot>>(raw.get()) {
579            return Ok(Self::SlimEnvelope {
580                timestamp: envelope.timestamp,
581                instrument_ticker: envelope.instrument_ticker,
582            });
583        }
584
585        serde_json::from_str(raw.get())
586            .map(Self::Ticker)
587            .map_err(serde::de::Error::custom)
588    }
589}
590
591impl DeriveTickerData {
592    /// Returns the ticker timestamp in UNIX milliseconds.
593    #[must_use]
594    pub const fn timestamp(&self) -> i64 {
595        match self {
596            Self::Envelope { timestamp, .. } => *timestamp,
597            Self::SlimEnvelope { timestamp, .. } => *timestamp,
598            Self::Ticker(ticker) => ticker.timestamp,
599        }
600    }
601
602    /// Returns the ticker instrument name.
603    #[must_use]
604    pub fn instrument_name(&self) -> &Ustr {
605        match self {
606            Self::Envelope {
607                instrument_ticker, ..
608            } => &instrument_ticker.instrument_name,
609            Self::SlimEnvelope {
610                instrument_ticker, ..
611            } => &instrument_ticker.instrument_name,
612            Self::Ticker(ticker) => &ticker.instrument_name,
613        }
614    }
615
616    /// Returns the best ask price.
617    #[must_use]
618    pub fn best_ask_price(&self) -> Decimal {
619        match self {
620            Self::Envelope {
621                instrument_ticker, ..
622            } => instrument_ticker.best_ask_price,
623            Self::SlimEnvelope {
624                instrument_ticker, ..
625            } => instrument_ticker.best_ask_price,
626            Self::Ticker(ticker) => ticker.best_ask_price,
627        }
628    }
629
630    /// Returns the best bid price.
631    #[must_use]
632    pub fn best_bid_price(&self) -> Decimal {
633        match self {
634            Self::Envelope {
635                instrument_ticker, ..
636            } => instrument_ticker.best_bid_price,
637            Self::SlimEnvelope {
638                instrument_ticker, ..
639            } => instrument_ticker.best_bid_price,
640            Self::Ticker(ticker) => ticker.best_bid_price,
641        }
642    }
643
644    /// Returns the best ask amount.
645    #[must_use]
646    pub fn best_ask_amount(&self) -> Decimal {
647        match self {
648            Self::Envelope {
649                instrument_ticker, ..
650            } => instrument_ticker.best_ask_amount,
651            Self::SlimEnvelope {
652                instrument_ticker, ..
653            } => instrument_ticker.best_ask_amount,
654            Self::Ticker(ticker) => ticker.best_ask_amount,
655        }
656    }
657
658    /// Returns the best bid amount.
659    #[must_use]
660    pub fn best_bid_amount(&self) -> Decimal {
661        match self {
662            Self::Envelope {
663                instrument_ticker, ..
664            } => instrument_ticker.best_bid_amount,
665            Self::SlimEnvelope {
666                instrument_ticker, ..
667            } => instrument_ticker.best_bid_amount,
668            Self::Ticker(ticker) => ticker.best_bid_amount,
669        }
670    }
671
672    /// Returns the current mark price.
673    #[must_use]
674    pub fn mark_price(&self) -> Decimal {
675        match self {
676            Self::Envelope {
677                instrument_ticker, ..
678            } => instrument_ticker.mark_price,
679            Self::SlimEnvelope {
680                instrument_ticker, ..
681            } => instrument_ticker.mark_price,
682            Self::Ticker(ticker) => ticker.mark_price,
683        }
684    }
685
686    /// Returns the current index price.
687    #[must_use]
688    pub fn index_price(&self) -> Decimal {
689        match self {
690            Self::Envelope {
691                instrument_ticker, ..
692            } => instrument_ticker.index_price,
693            Self::SlimEnvelope {
694                instrument_ticker, ..
695            } => instrument_ticker.index_price,
696            Self::Ticker(ticker) => ticker.index_price,
697        }
698    }
699
700    /// Returns the current funding rate when the ticker carries one.
701    #[must_use]
702    pub fn funding_rate(&self) -> Option<Decimal> {
703        match self {
704            Self::Envelope {
705                instrument_ticker, ..
706            } => instrument_ticker
707                .perp_details
708                .as_ref()
709                .map(|perp| perp.funding_rate),
710            Self::SlimEnvelope {
711                instrument_ticker, ..
712            } => instrument_ticker.funding_rate,
713            Self::Ticker(ticker) => ticker.perp_details.as_ref().map(|perp| perp.funding_rate),
714        }
715    }
716
717    /// Returns option pricing fields when the ticker carries them.
718    #[must_use]
719    pub fn option_pricing(&self) -> Option<&DeriveOptionPricing> {
720        match self {
721            Self::Envelope {
722                instrument_ticker, ..
723            } => instrument_ticker.option_pricing.as_ref(),
724            Self::SlimEnvelope {
725                instrument_ticker, ..
726            } => instrument_ticker.option_pricing.as_ref(),
727            Self::Ticker(ticker) => ticker.option_pricing.as_ref(),
728        }
729    }
730
731    /// Returns 24-hour aggregate statistics when the ticker carries them.
732    #[must_use]
733    pub fn stats(&self) -> Option<&DeriveAggregateTradingStats> {
734        match self {
735            Self::Envelope {
736                instrument_ticker, ..
737            } => instrument_ticker.stats.as_ref(),
738            Self::SlimEnvelope {
739                instrument_ticker, ..
740            } => instrument_ticker.stats.as_ref(),
741            Self::Ticker(ticker) => ticker.stats.as_ref(),
742        }
743    }
744
745    /// Fills compact ticker context that the venue omits from `ticker_slim`.
746    ///
747    /// # Errors
748    ///
749    /// Returns an error if a compact ticker is received on an invalid channel.
750    pub fn apply_channel_context(&mut self, channel: &str) -> Result<(), String> {
751        let Self::SlimEnvelope {
752            instrument_ticker, ..
753        } = self
754        else {
755            return Ok(());
756        };
757
758        if !instrument_ticker.instrument_name.is_empty() {
759            return Ok(());
760        }
761
762        let instrument_name = ticker_instrument_name_from_channel(channel)
763            .ok_or_else(|| format!("invalid Derive ticker channel `{channel}`"))?;
764        instrument_ticker.instrument_name = Ustr::from(instrument_name);
765        Ok(())
766    }
767
768    /// Returns the Nautilus instrument ID for this Derive symbol.
769    ///
770    /// # Errors
771    ///
772    /// Returns an error if the venue symbol is empty or contains only whitespace.
773    pub fn instrument_id(&self) -> anyhow::Result<InstrumentId> {
774        format_instrument_id(self.instrument_name())
775    }
776}
777
778/// Channel-tagged ticker update.
779#[derive(Debug, Clone)]
780pub struct DeriveTickerMsg {
781    /// Channel that produced the update.
782    pub channel: Ustr,
783    /// Parsed ticker data.
784    pub data: DeriveTickerData,
785}
786
787/// Typed public market data update parsed from a Derive subscription frame.
788#[derive(Debug, Clone)]
789pub enum DerivePublicWsData {
790    /// Order book snapshot update.
791    Orderbook(DeriveOrderbookMsg),
792    /// Public trades update.
793    Trades(DeriveTradesMsg),
794    /// Ticker update.
795    Ticker(Box<DeriveTickerMsg>),
796}
797
798/// Inbound frame discriminated by whether it carries an `id` (response to a
799/// client request) or a `method` (server-initiated notification).
800#[derive(Debug, Clone)]
801pub enum DeriveWsFrame {
802    /// JSON-RPC response correlated with an outbound request `id`.
803    Response {
804        /// Echoed request id.
805        id: u64,
806        /// Result payload when the venue accepted the request.
807        result: Option<Value>,
808        /// Error payload when the venue rejected the request.
809        error: Option<JsonRpcError>,
810    },
811    /// Server-initiated subscription update.
812    Subscription(WsSubscriptionPayload),
813    /// JSON-RPC error whose null or absent id cannot identify a pending request.
814    UncorrelatedError(JsonRpcError),
815    /// Frame we could decode as JSON but did not recognize; surfaced so logs
816    /// can flag unknown server-initiated messages without dropping silently.
817    Unknown(Value),
818}
819
820#[derive(Debug)]
821pub(super) enum ParsedFrame<R> {
822    Response {
823        id: u64,
824        result: Option<R>,
825        error: Option<JsonRpcError>,
826    },
827    Subscription(WsSubscriptionPayload),
828    UncorrelatedError(JsonRpcError),
829    Unknown(Value),
830}
831
832/// Single-pass deserialize target for an inbound frame.
833///
834/// `params` is captured as a raw [`RawValue`] span rather than eagerly decoded:
835/// it is only parsed into a [`WsSubscriptionPayload`] once the method check
836/// confirms a subscription, so a non-subscription notification carrying an
837/// unrelated `params` object still classifies as `Unknown` instead of failing
838/// the frame parse. Financial responses retain raw JSON tokens until typed decoding.
839#[derive(Debug, Deserialize)]
840#[serde(bound(deserialize = "R: Deserialize<'de>"))]
841struct InboundFrame<R> {
842    #[serde(default)]
843    id: Option<u64>,
844    #[serde(default)]
845    method: Option<Ustr>,
846    #[serde(default)]
847    result: Option<R>,
848    #[serde(default)]
849    error: Option<JsonRpcError>,
850    #[serde(default)]
851    params: Option<Box<RawValue>>,
852}
853
854impl DeriveWsFrame {
855    /// Parses a raw text frame into the discriminated [`DeriveWsFrame`].
856    ///
857    /// Returns the original JSON error when the bytes are not valid JSON;
858    /// callers log and drop the frame in that case.
859    ///
860    /// # Errors
861    ///
862    /// Returns an error if `text` is not valid JSON.
863    pub fn parse(text: &str) -> serde_json::Result<Self> {
864        match parse_frame(text)? {
865            ParsedFrame::Response { id, result, error } => Ok(Self::Response { id, result, error }),
866            ParsedFrame::Subscription(payload) => Ok(Self::Subscription(payload)),
867            ParsedFrame::UncorrelatedError(error) => Ok(Self::UncorrelatedError(error)),
868            ParsedFrame::Unknown(value) => Ok(Self::Unknown(value)),
869        }
870    }
871}
872
873impl ParsedFrame<Box<RawValue>> {
874    pub(super) fn parse(text: &str) -> serde_json::Result<Self> {
875        parse_frame(text)
876    }
877}
878
879fn parse_frame<R: serde::de::DeserializeOwned>(text: &str) -> serde_json::Result<ParsedFrame<R>> {
880    let frame: InboundFrame<R> = serde_json::from_str(text)?;
881
882    if let Some(id) = frame.id {
883        return Ok(ParsedFrame::Response {
884            id,
885            result: frame.result,
886            error: frame.error,
887        });
888    }
889
890    if frame
891        .method
892        .as_ref()
893        .is_some_and(|method| method.as_str() == "subscription")
894        && let Some(params) = frame.params
895    {
896        let payload: WsSubscriptionPayload = serde_json::from_str(params.get())?;
897        return Ok(ParsedFrame::Subscription(payload));
898    }
899
900    if let Some(error) = frame.error {
901        return Ok(ParsedFrame::UncorrelatedError(error));
902    }
903
904    // Unrecognized frame: re-parse into a `Value` for diagnostic logging.
905    // The live feed only sends responses and subscription notifications, so
906    // this second parse never runs on a hot path.
907    Ok(ParsedFrame::Unknown(serde_json::from_str(text)?))
908}
909
910/// Formats the topic for the public `ticker_slim.{instrument_name}.{interval}` channel.
911///
912/// `interval` is the millisecond cadence the venue exposes (e.g. `"100"`,
913/// `"1000"`). The function does not validate the value; the venue rejects
914/// unsupported intervals on subscribe.
915#[must_use]
916pub fn ticker_channel(instrument_name: &str, interval: &str) -> String {
917    format!("ticker_slim.{instrument_name}.{interval}")
918}
919
920fn ticker_instrument_name_from_channel(channel: &str) -> Option<&str> {
921    let rest = channel
922        .strip_prefix("ticker_slim.")
923        .or_else(|| channel.strip_prefix("ticker."))?;
924    let (instrument_name, _) = rest.rsplit_once('.')?;
925    (!instrument_name.is_empty()).then_some(instrument_name)
926}
927
928/// Formats the topic for `orderbook.{instrument_name}.{group}.{depth}`.
929#[must_use]
930pub fn orderbook_channel(instrument_name: &str, group: &str, depth: &str) -> String {
931    format!("orderbook.{instrument_name}.{group}.{depth}")
932}
933
934/// Formats the topic for `trades.{instrument_type}.{currency}`.
935#[must_use]
936pub fn trades_channel(instrument_type: &str, currency: &str) -> String {
937    format!("trades.{instrument_type}.{currency}")
938}
939
940/// Formats the topic for the private `{subaccount_id}.orders` channel.
941#[must_use]
942pub fn orders_channel(subaccount_id: u64) -> String {
943    format!("{subaccount_id}.orders")
944}
945
946/// Formats the topic for the private `{subaccount_id}.trades` channel.
947#[must_use]
948pub fn private_trades_channel(subaccount_id: u64) -> String {
949    format!("{subaccount_id}.trades")
950}
951
952/// Formats the topic for the private `{subaccount_id}.balances` channel.
953#[must_use]
954pub fn balances_channel(subaccount_id: u64) -> String {
955    format!("{subaccount_id}.balances")
956}
957
958/// JSON-RPC method names exchanged on the Derive WebSocket transport.
959///
960/// The `private/*` trading methods mirror the REST endpoints exactly: the
961/// signed EIP-712 params built in [`crate::http::query`] and the result
962/// envelopes in [`crate::http::models`] are reused verbatim over the
963/// WebSocket. The session is authorized once via `PUBLIC_LOGIN`; no
964/// per-request auth headers are sent.
965pub mod methods {
966    /// Authenticated session login. Params: [`super::WsLoginParams`].
967    pub const PUBLIC_LOGIN: &str = "public/login";
968    /// Subscribe to a list of channels. Params: [`super::WsSubscribeParams`].
969    pub const PUBLIC_SUBSCRIBE: &str = "subscribe";
970    /// Unsubscribe from a list of channels. Params: [`super::WsUnsubscribeParams`].
971    pub const PUBLIC_UNSUBSCRIBE: &str = "unsubscribe";
972    /// Submit a signed order. Params: [`crate::http::query::DeriveOrderParams`].
973    pub const PRIVATE_ORDER: &str = "private/order";
974    /// Submit a signed trigger order. Params:
975    /// [`crate::http::query::DeriveTriggerOrderParams`].
976    pub const PRIVATE_TRIGGER_ORDER: &str = PRIVATE_ORDER;
977    /// Cancel a single order. Params: [`crate::http::query::DeriveCancelParams`].
978    pub const PRIVATE_CANCEL: &str = "private/cancel";
979    /// Cancel every open order for one instrument. Params:
980    /// [`crate::http::query::DeriveCancelByInstrumentParams`].
981    pub const PRIVATE_CANCEL_BY_INSTRUMENT: &str = "private/cancel_by_instrument";
982    /// Cancel a single trigger order. Params:
983    /// [`crate::http::query::DeriveCancelTriggerOrderParams`].
984    pub const PRIVATE_CANCEL_TRIGGER_ORDER: &str = "private/cancel_trigger_order";
985    /// Cancel orders by label. Params:
986    /// [`crate::http::query::DeriveCancelByLabelParams`].
987    pub const PRIVATE_CANCEL_BY_LABEL: &str = "private/cancel_by_label";
988    /// List untriggered trigger orders. Params:
989    /// [`crate::http::query::DeriveGetTriggerOrdersParams`].
990    pub const PRIVATE_GET_TRIGGER_ORDERS: &str = "private/get_trigger_orders";
991    /// Cancel every open order on the subaccount, optionally scoped to an
992    /// instrument. Params: [`crate::http::query::DeriveCancelAllParams`].
993    pub const PRIVATE_CANCEL_ALL: &str = "private/cancel_all";
994    /// Sets wallet-wide cancel-on-disconnect. Params: [`super::WsSetCancelOnDisconnectParams`].
995    pub const PRIVATE_SET_CANCEL_ON_DISCONNECT: &str = "private/set_cancel_on_disconnect";
996    /// Atomically cancel one order and submit a replacement. Params:
997    /// [`crate::http::query::DeriveReplaceParams`].
998    pub const PRIVATE_REPLACE: &str = "private/replace";
999}
1000
1001#[cfg(test)]
1002mod tests {
1003    use rstest::rstest;
1004    use serde_json::json;
1005
1006    use super::*;
1007    use crate::http::models::JsonRpcRequest;
1008
1009    #[rstest]
1010    fn test_ticker_channel_joins_with_dots() {
1011        assert_eq!(
1012            ticker_channel("ETH-PERP", "1000"),
1013            "ticker_slim.ETH-PERP.1000",
1014        );
1015        assert_eq!(
1016            ticker_channel("BTC-20260627-100000-C", "100"),
1017            "ticker_slim.BTC-20260627-100000-C.100",
1018        );
1019    }
1020
1021    #[rstest]
1022    fn test_orderbook_channel_joins_with_dots() {
1023        assert_eq!(
1024            orderbook_channel("ETH-PERP", "1", "10"),
1025            "orderbook.ETH-PERP.1.10",
1026        );
1027    }
1028
1029    #[rstest]
1030    fn test_trades_channel_joins_with_dots() {
1031        assert_eq!(trades_channel("perp", "ETH"), "trades.perp.ETH");
1032    }
1033
1034    #[rstest]
1035    #[case(0_u64, "0.orders", "0.trades", "0.balances")]
1036    #[case(1_u64, "1.orders", "1.trades", "1.balances")]
1037    #[case(30769_u64, "30769.orders", "30769.trades", "30769.balances")]
1038    fn test_private_channel_formatters_emit_subaccount_prefix(
1039        #[case] subaccount: u64,
1040        #[case] expected_orders: &str,
1041        #[case] expected_trades: &str,
1042        #[case] expected_balances: &str,
1043    ) {
1044        assert_eq!(orders_channel(subaccount), expected_orders);
1045        assert_eq!(private_trades_channel(subaccount), expected_trades);
1046        assert_eq!(balances_channel(subaccount), expected_balances);
1047    }
1048
1049    #[rstest]
1050    fn test_ws_channel_formats_known_topics() {
1051        assert_eq!(
1052            DeriveWsChannel::ticker_slim("ETH-PERP", DeriveTickerInterval::Ms1000).to_string(),
1053            "ticker_slim.ETH-PERP.1000",
1054        );
1055        assert_eq!(
1056            DeriveWsChannel::orderbook(
1057                "ETH-PERP",
1058                DeriveOrderbookGroup::G1,
1059                DeriveOrderbookDepth::D10,
1060            )
1061            .to_string(),
1062            "orderbook.ETH-PERP.1.10",
1063        );
1064        assert_eq!(
1065            DeriveWsChannel::trades(DeriveInstrumentType::Perp, "ETH").to_string(),
1066            "trades.perp.ETH",
1067        );
1068        assert_eq!(DeriveWsChannel::orders(30769).to_string(), "30769.orders");
1069        assert_eq!(
1070            DeriveWsChannel::private_trades(30769).to_string(),
1071            "30769.trades",
1072        );
1073        assert_eq!(
1074            DeriveWsChannel::balances(30769).to_string(),
1075            "30769.balances",
1076        );
1077    }
1078
1079    #[rstest]
1080    fn test_ws_channel_deserializes_known_and_raw_topics() {
1081        let ticker: DeriveWsChannel =
1082            serde_json::from_value(json!("ticker_slim.ETH.TEST-PERP.1000")).unwrap();
1083        let orderbook: DeriveWsChannel =
1084            serde_json::from_value(json!("orderbook.ETH.TEST-PERP.1.10")).unwrap();
1085        let private_trades: DeriveWsChannel =
1086            serde_json::from_value(json!("30769.trades")).unwrap();
1087        let raw: DeriveWsChannel = serde_json::from_value(json!("trades.ETH-USDC")).unwrap();
1088
1089        assert_eq!(
1090            ticker,
1091            DeriveWsChannel::ticker_slim("ETH.TEST-PERP", "1000"),
1092        );
1093        assert_eq!(
1094            orderbook,
1095            DeriveWsChannel::orderbook("ETH.TEST-PERP", "1", "10"),
1096        );
1097        assert_eq!(private_trades, DeriveWsChannel::private_trades(30769));
1098        assert_eq!(raw, DeriveWsChannel::Raw("trades.ETH-USDC".to_string()));
1099    }
1100
1101    #[rstest]
1102    fn test_ws_channel_uses_typed_known_topic_fields() {
1103        let ticker = DeriveWsChannel::from_topic("ticker_slim.ETH-PERP.1000");
1104        let orderbook = DeriveWsChannel::from_topic("orderbook.ETH-PERP.1.10");
1105        let trades = DeriveWsChannel::from_topic("trades.perp.ETH");
1106
1107        match ticker {
1108            DeriveWsChannel::TickerSlim {
1109                instrument_name,
1110                interval,
1111            } => {
1112                assert_eq!(instrument_name, "ETH-PERP");
1113                assert_eq!(interval, DeriveTickerInterval::Ms1000);
1114            }
1115            other => panic!("expected TickerSlim, was {other:?}"),
1116        }
1117
1118        match orderbook {
1119            DeriveWsChannel::Orderbook {
1120                instrument_name,
1121                group,
1122                depth,
1123            } => {
1124                assert_eq!(instrument_name, "ETH-PERP");
1125                assert_eq!(group, DeriveOrderbookGroup::G1);
1126                assert_eq!(depth, DeriveOrderbookDepth::D10);
1127            }
1128            other => panic!("expected Orderbook, was {other:?}"),
1129        }
1130
1131        match trades {
1132            DeriveWsChannel::Trades {
1133                instrument_type,
1134                currency,
1135            } => {
1136                assert_eq!(instrument_type, DeriveInstrumentType::Perp);
1137                assert_eq!(currency, "ETH");
1138            }
1139            other => panic!("expected Trades, was {other:?}"),
1140        }
1141    }
1142
1143    #[rstest]
1144    fn test_subscribe_request_serializes_as_jsonrpc_envelope() {
1145        let req = JsonRpcRequest::new(
1146            1,
1147            methods::PUBLIC_SUBSCRIBE,
1148            WsSubscribeParams {
1149                channels: vec![DeriveWsChannel::ticker_slim("ETH-PERP", "1000")],
1150            },
1151        );
1152
1153        let wire = serde_json::to_value(&req).unwrap();
1154        assert_eq!(wire["jsonrpc"], "2.0");
1155        assert_eq!(wire["id"], 1);
1156        assert_eq!(wire["method"], "subscribe");
1157        assert_eq!(wire["params"]["channels"][0], "ticker_slim.ETH-PERP.1000");
1158    }
1159
1160    #[rstest]
1161    fn test_ws_request_params_preserve_jsonrpc_wire_output() {
1162        let login = JsonRpcRequest::new(
1163            1,
1164            methods::PUBLIC_LOGIN,
1165            WsRequestParams::from(WsLoginParams {
1166                wallet: "0xWALLET".to_string(),
1167                timestamp: "1700000000000".to_string(),
1168                signature: SecretString::from("0xSIG"),
1169            }),
1170        );
1171
1172        let subscribe = JsonRpcRequest::new(
1173            2,
1174            methods::PUBLIC_SUBSCRIBE,
1175            WsRequestParams::from(WsSubscribeParams {
1176                channels: vec![DeriveWsChannel::ticker_slim("ETH-PERP", "1000")],
1177            }),
1178        );
1179
1180        let unsubscribe = JsonRpcRequest::new(
1181            3,
1182            methods::PUBLIC_UNSUBSCRIBE,
1183            WsRequestParams::from(WsUnsubscribeParams {
1184                channels: vec![DeriveWsChannel::ticker_slim("ETH-PERP", "1000")],
1185            }),
1186        );
1187
1188        assert_eq!(
1189            serde_json::to_string(&login).unwrap(),
1190            concat!(
1191                r#"{"jsonrpc":"2.0","id":1,"method":"public/login","params":{"#,
1192                r#""wallet":"0xWALLET","timestamp":"1700000000000","signature":"0xSIG"}}"#,
1193            ),
1194        );
1195        assert_eq!(
1196            serde_json::to_string(&subscribe).unwrap(),
1197            concat!(
1198                r#"{"jsonrpc":"2.0","id":2,"method":"subscribe","params":{"#,
1199                r#""channels":["ticker_slim.ETH-PERP.1000"]}}"#,
1200            ),
1201        );
1202        assert_eq!(
1203            serde_json::to_string(&unsubscribe).unwrap(),
1204            concat!(
1205                r#"{"jsonrpc":"2.0","id":3,"method":"unsubscribe","params":{"#,
1206                r#""channels":["ticker_slim.ETH-PERP.1000"]}}"#,
1207            ),
1208        );
1209    }
1210
1211    #[rstest]
1212    fn test_ws_response_results_decode_known_shapes() {
1213        let login: WsLoginResult = serde_json::from_value(json!([30769, 42])).unwrap();
1214        let subscribe: WsSubscribeResult = serde_json::from_value(json!({
1215            "current_subscriptions": ["ticker_slim.ETH-PERP.1000"],
1216            "status": {"ticker_slim.ETH-PERP.1000": "ok"},
1217        }))
1218        .unwrap();
1219        let unsubscribe: WsUnsubscribeResult = serde_json::from_value(
1220            json!({"status": {"ticker_slim.ETH-PERP.1000": "ok"}, "remaining_subscriptions": []}),
1221        )
1222        .unwrap();
1223
1224        assert_eq!(login, vec![30769, 42]);
1225        assert_eq!(
1226            subscribe.channels,
1227            vec![DeriveWsChannel::ticker_slim("ETH-PERP", "1000")],
1228        );
1229        assert_eq!(
1230            unsubscribe.status[&DeriveWsChannel::ticker_slim("ETH-PERP", "1000")],
1231            "ok"
1232        );
1233        assert!(unsubscribe.remaining_subscriptions.is_empty());
1234    }
1235
1236    #[rstest]
1237    fn test_subscribe_result_decodes_recorded_venue_ack() {
1238        let ack: Value =
1239            serde_json::from_str(include_str!("../../test_data/spot/ws_subscribe_ack.json"))
1240                .unwrap();
1241        let result: WsSubscribeResult =
1242            serde_json::from_value(ack["result"].clone()).expect("subscribe ack parses");
1243
1244        assert!(
1245            result
1246                .channels
1247                .contains(&DeriveWsChannel::ticker_slim("ETH-USDC", "1000"))
1248        );
1249        assert_eq!(
1250            result
1251                .status
1252                .get(&DeriveWsChannel::orderbook("ETH-USDC", "1", "10"))
1253                .map(|status| status.as_str()),
1254            Some("ok"),
1255        );
1256    }
1257
1258    #[rstest]
1259    fn test_login_params_round_trip() {
1260        let params = WsLoginParams {
1261            wallet: "0xWALLET".to_string(),
1262            timestamp: "1700000000000".to_string(),
1263            signature: SecretString::from("0xDEAD"),
1264        };
1265
1266        let debug = format!("{params:?}");
1267        let wire = serde_json::to_value(&params).unwrap();
1268
1269        assert_eq!(wire["wallet"], "0xWALLET");
1270        assert_eq!(wire["timestamp"], "1700000000000");
1271        assert_eq!(wire["signature"], "0xDEAD");
1272        assert!(debug.contains(REDACTED));
1273        assert!(!debug.contains("0xDEAD"));
1274        let back: WsLoginParams = serde_json::from_value(wire).unwrap();
1275        assert_eq!(back, params);
1276    }
1277
1278    #[rstest]
1279    fn test_parse_response_with_result() {
1280        let text = json!({"id": 42, "result": {"ok": true}}).to_string();
1281        let frame = DeriveWsFrame::parse(&text).unwrap();
1282        match frame {
1283            DeriveWsFrame::Response { id, result, error } => {
1284                assert_eq!(id, 42);
1285                assert_eq!(result, Some(json!({"ok": true})));
1286                assert!(error.is_none());
1287            }
1288            other => panic!("expected Response, was {other:?}"),
1289        }
1290    }
1291
1292    #[rstest]
1293    fn test_parse_response_with_error_payload() {
1294        let text = json!({
1295            "id": 7,
1296            "error": {"code": -32602, "message": "bad params", "data": {"field": "channels"}},
1297        })
1298        .to_string();
1299        let frame = DeriveWsFrame::parse(&text).unwrap();
1300        match frame {
1301            DeriveWsFrame::Response { id, result, error } => {
1302                assert_eq!(id, 7);
1303                assert!(result.is_none());
1304                let err = error.expect("error present");
1305                assert_eq!(err.code, -32602);
1306                assert_eq!(err.data, Some(json!({"field": "channels"})));
1307            }
1308            other => panic!("expected Response, was {other:?}"),
1309        }
1310    }
1311
1312    #[rstest]
1313    fn test_parse_subscription_notification() {
1314        let text = json!({
1315            "method": "subscription",
1316            "params": {
1317                "channel": "ticker.ETH-PERP.1000",
1318                "data": {"instrument_name": "ETH-PERP", "mark_price": "3500.5"},
1319            },
1320        })
1321        .to_string();
1322        let frame = DeriveWsFrame::parse(&text).unwrap();
1323        match frame {
1324            DeriveWsFrame::Subscription(payload) => {
1325                assert_eq!(payload.channel, "ticker.ETH-PERP.1000");
1326                let data: Value = serde_json::from_str(payload.data.get()).unwrap();
1327                assert_eq!(data["mark_price"], "3500.5");
1328            }
1329            other => panic!("expected Subscription, was {other:?}"),
1330        }
1331    }
1332
1333    #[rstest]
1334    fn test_parse_unknown_frame_preserves_value() {
1335        let text = json!({"hello": "world"}).to_string();
1336        let frame = DeriveWsFrame::parse(&text).unwrap();
1337        match frame {
1338            DeriveWsFrame::Unknown(v) => {
1339                assert_eq!(v["hello"], "world");
1340                assert!(v.get("id").is_none(), "unknown frame must not carry id");
1341                let method = v.get("method").and_then(Value::as_str);
1342                assert_ne!(method, Some("subscription"));
1343            }
1344            other => panic!("expected Unknown, was {other:?}"),
1345        }
1346    }
1347
1348    #[rstest]
1349    fn test_parse_null_id_error_preserves_structured_error() {
1350        let text = json!({
1351            "id": null,
1352            "error": {
1353                "code": -32700,
1354                "message": "Parse error",
1355                "data": "invalid JSON",
1356            },
1357        })
1358        .to_string();
1359        let frame = DeriveWsFrame::parse(&text).unwrap();
1360
1361        match frame {
1362            DeriveWsFrame::UncorrelatedError(error) => {
1363                assert_eq!(error.code, -32700);
1364                assert_eq!(error.message, "Parse error");
1365                assert_eq!(error.data, Some(json!("invalid JSON")));
1366            }
1367            other => panic!("expected UncorrelatedError, was {other:?}"),
1368        }
1369    }
1370
1371    #[rstest]
1372    fn test_parse_non_subscription_notification_with_params_is_unknown() {
1373        // A non-subscription notification that carries an unrelated `params`
1374        // object must classify as Unknown, not fail the frame parse: the
1375        // params shape is only checked once the method confirms a subscription.
1376        let text = json!({"method": "heartbeat", "params": {"interval": 30}}).to_string();
1377        let frame = DeriveWsFrame::parse(&text).unwrap();
1378        match frame {
1379            DeriveWsFrame::Unknown(v) => {
1380                assert_eq!(v["method"], "heartbeat");
1381                assert_eq!(v["params"]["interval"], 30);
1382            }
1383            other => panic!("expected Unknown, was {other:?}"),
1384        }
1385    }
1386
1387    #[rstest]
1388    fn test_parse_response_with_both_result_and_error_prefers_error() {
1389        // FeedHandler dispatch treats error as winning when both are present.
1390        let text = json!({
1391            "id": 11,
1392            "result": {"should_not_win": true},
1393            "error": {"code": -1, "message": "wins"},
1394        })
1395        .to_string();
1396        let frame = DeriveWsFrame::parse(&text).unwrap();
1397        match frame {
1398            DeriveWsFrame::Response { id, result, error } => {
1399                assert_eq!(id, 11);
1400                assert!(result.is_some(), "result is preserved on the frame");
1401                let err = error.expect("error present");
1402                assert_eq!(err.code, -1);
1403                assert_eq!(err.message, "wins");
1404            }
1405            other => panic!("expected Response, was {other:?}"),
1406        }
1407    }
1408
1409    #[rstest]
1410    fn test_parse_rejects_malformed_json() {
1411        let err = DeriveWsFrame::parse("{not json").expect_err("must reject");
1412        // Pin the variant so a future refactor swallowing parse errors into Ok(Unknown) fails.
1413        assert_eq!(err.classify(), serde_json::error::Category::Syntax);
1414    }
1415
1416    #[rstest]
1417    fn test_unsubscribe_params_round_trip() {
1418        let params = WsUnsubscribeParams {
1419            channels: vec![
1420                DeriveWsChannel::ticker_slim("ETH-PERP", "1000"),
1421                DeriveWsChannel::ticker_slim("BTC-PERP", "100"),
1422            ],
1423        };
1424
1425        let wire = serde_json::to_value(&params).unwrap();
1426        assert_eq!(wire["channels"][0], "ticker_slim.ETH-PERP.1000");
1427        assert_eq!(wire["channels"][1], "ticker_slim.BTC-PERP.100");
1428        let back: WsUnsubscribeParams = serde_json::from_value(wire).unwrap();
1429        assert_eq!(back, params);
1430    }
1431
1432    #[rstest]
1433    fn test_private_orders_subscription_data_decodes_single_and_array_payloads() {
1434        let order: Value = serde_json::from_str(include_str!(
1435            "../../test_data/perps/http_order_eth_partially_filled.json"
1436        ))
1437        .unwrap();
1438        let single: DeriveOrdersSubscriptionData = serde_json::from_value(order.clone()).unwrap();
1439        let array: DeriveOrdersSubscriptionData =
1440            serde_json::from_value(json!([order, {"not": "an order"}])).unwrap();
1441
1442        assert_eq!(single.orders.len(), 1);
1443        assert_eq!(single.orders[0].order_id, "abc-123");
1444        assert_eq!(array.orders.len(), 1);
1445        assert_eq!(array.orders[0].order_id, "abc-123");
1446    }
1447
1448    #[rstest]
1449    fn test_private_trades_subscription_data_decodes_single_and_array_payloads() {
1450        let trade: Value = serde_json::from_str(include_str!(
1451            "../../test_data/perps/http_private_trade_eth.json"
1452        ))
1453        .unwrap();
1454        let single: DeriveTradesSubscriptionData = serde_json::from_value(trade.clone()).unwrap();
1455        let array: DeriveTradesSubscriptionData =
1456            serde_json::from_value(json!([trade, {"not": "a trade"}])).unwrap();
1457
1458        assert_eq!(single.trades.len(), 1);
1459        assert_eq!(single.trades[0].trade_id, "trade-xyz");
1460        assert_eq!(array.trades.len(), 1);
1461        assert_eq!(array.trades[0].trade_id, "trade-xyz");
1462    }
1463}
nautilus_derive/websocket/messages.rs

nautilus_derive/websocket/
messages.rs
