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

nautilus_derive/websocket/
messages.rs
