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

nautilus_derive/websocket/
messages.rs
