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

nautilus_derive/websocket/
messages.rs
