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

nautilus_derive/websocket/
messages.rs
