dydx/indexer/sock/
messages.rs

1use crate::indexer::types::{
2    CandleResponse as CandlesInitialMessageContents, CandleResponseObject as Candle,
3    HeightResponse as BlockHeightInitialMessageContents,
4    OrderBookResponseObject as OrdersInitialMessageContents,
5    OrderResponseObject as OrderMessageObject,
6    ParentSubaccountResponseObject as ParentSubaccountMessageObject,
7    PerpetualMarketResponse as MarketsInitialMessageContents,
8    SubaccountResponseObject as SubaccountMessageObject,
9    TradeResponse as TradesInitialMessageContents, *,
10};
11use bigdecimal::BigDecimal;
12use chrono::{DateTime, Utc};
13use serde::Deserialize;
14use serde_json::{json, Value};
15use std::collections::HashMap;
16use tokio_tungstenite::tungstenite::protocol::Message;
17
18/// Feed subscription options
19/// Respective ticker is required for `Orders`, `Trades`, `Candles`
20#[derive(Debug, Clone, Hash, Eq, PartialEq)]
21pub enum Subscription {
22    /// Subaccounts.
23    Subaccounts(Subaccount),
24    /// Orders.
25    Orders(Ticker),
26    /// Trades.
27    Trades(Ticker),
28    /// Markets.
29    Markets,
30    /// Candles.
31    Candles(Ticker, CandleResolution),
32    /// Parent subaccounts.
33    ParentSubaccounts(ParentSubaccount),
34    /// Block height.
35    BlockHeight,
36}
37
38impl Subscription {
39    pub(crate) fn sub_message(&self, batched: bool) -> Message {
40        match self {
41            Self::Subaccounts(ref subacc) => subaccounts::sub_message(subacc, batched),
42            Self::Markets => markets::sub_message(batched),
43            Self::Orders(ref ticker) => orders::sub_message(ticker, batched),
44            Self::Trades(ref ticker) => trades::sub_message(ticker, batched),
45            Self::Candles(ref ticker, ref res) => candles::sub_message(ticker, res, batched),
46            Self::ParentSubaccounts(ref subacc) => parent_subaccounts::sub_message(subacc, batched),
47            Self::BlockHeight => block_height::sub_message(batched),
48        }
49    }
50
51    pub(crate) fn unsub_message(&self) -> Message {
52        match self {
53            Self::Subaccounts(ref subacc) => subaccounts::unsub_message(subacc),
54            Self::Markets => markets::unsub_message(),
55            Self::Orders(ref ticker) => orders::unsub_message(ticker),
56            Self::Trades(ref ticker) => trades::unsub_message(ticker),
57            Self::Candles(ref ticker, ref res) => candles::unsub_message(ticker, res),
58            Self::ParentSubaccounts(ref subacc) => parent_subaccounts::unsub_message(subacc),
59            Self::BlockHeight => block_height::unsub_message(),
60        }
61    }
62}
63
64struct MessageFormatter {}
65
66impl MessageFormatter {
67    pub(crate) fn sub_message(channel: &str, fields: Value) -> Message {
68        let message = json!({
69            "type": "subscribe",
70            "channel": channel,
71        });
72        Self::message(fields, message)
73    }
74
75    pub(crate) fn unsub_message(channel: &str, fields: Value) -> Message {
76        let message = json!({
77            "type": "unsubscribe",
78            "channel": channel,
79        });
80        Self::message(fields, message)
81    }
82
83    fn message(mut message: Value, fields: Value) -> Message {
84        if let Value::Object(ref mut map) = message {
85            if let Value::Object(fields) = fields {
86                map.extend(fields);
87            }
88        }
89        Message::Text(message.to_string().into())
90    }
91}
92
93pub(crate) mod subaccounts {
94    use super::{json, Message, MessageFormatter, Subaccount};
95    pub(crate) const CHANNEL: &str = "v4_subaccounts";
96
97    pub(crate) fn sub_message(subacc: &Subaccount, batched: bool) -> Message {
98        let address = &subacc.address;
99        let number = &subacc.number;
100        MessageFormatter::sub_message(
101            CHANNEL,
102            json!({"id": format!("{address}/{number}"), "batched": batched}),
103        )
104    }
105
106    pub(crate) fn unsub_message(subacc: &Subaccount) -> Message {
107        let address = &subacc.address;
108        let number = &subacc.number;
109        MessageFormatter::unsub_message(CHANNEL, json!({"id": format!("{address}/{number}")}))
110    }
111}
112
113pub(crate) mod parent_subaccounts {
114    use super::{json, Message, MessageFormatter, ParentSubaccount};
115    pub(crate) const CHANNEL: &str = "v4_parent_subaccounts";
116
117    pub(crate) fn sub_message(subacc: &ParentSubaccount, batched: bool) -> Message {
118        let address = &subacc.address;
119        let number = &subacc.number;
120        MessageFormatter::sub_message(
121            CHANNEL,
122            json!({"id": format!("{address}/{number}"), "batched": batched}),
123        )
124    }
125
126    pub(crate) fn unsub_message(subacc: &ParentSubaccount) -> Message {
127        let address = &subacc.address;
128        let number = &subacc.number;
129        MessageFormatter::unsub_message(CHANNEL, json!({"id": format!("{address}/{number}")}))
130    }
131}
132
133pub(crate) mod orders {
134    use super::{json, Message, MessageFormatter, Ticker};
135    pub(crate) const CHANNEL: &str = "v4_orderbook";
136
137    pub(crate) fn sub_message(id: &Ticker, batched: bool) -> Message {
138        MessageFormatter::sub_message(CHANNEL, json!({"id": id, "batched": batched}))
139    }
140
141    pub(crate) fn unsub_message(id: &Ticker) -> Message {
142        MessageFormatter::unsub_message(CHANNEL, json!({"id": id}))
143    }
144}
145
146pub(crate) mod trades {
147    use super::{json, Message, MessageFormatter, Ticker};
148    pub(crate) const CHANNEL: &str = "v4_trades";
149
150    pub(crate) fn sub_message(id: &Ticker, batched: bool) -> Message {
151        MessageFormatter::sub_message(CHANNEL, json!({"id": id, "batched": batched}))
152    }
153
154    pub(crate) fn unsub_message(id: &Ticker) -> Message {
155        MessageFormatter::unsub_message(CHANNEL, json!({"id": id}))
156    }
157}
158
159pub(crate) mod markets {
160    use super::{json, Message, MessageFormatter};
161    pub const CHANNEL: &str = "v4_markets";
162
163    pub(crate) fn sub_message(batched: bool) -> Message {
164        MessageFormatter::sub_message(CHANNEL, json!({"batched": batched}))
165    }
166
167    pub(crate) fn unsub_message() -> Message {
168        MessageFormatter::unsub_message(CHANNEL, json!({}))
169    }
170}
171
172pub(crate) mod candles {
173    use super::{json, CandleResolution, Message, MessageFormatter, Ticker};
174    pub(crate) const CHANNEL: &str = "v4_candles";
175
176    pub(crate) fn sub_message(
177        id: &Ticker,
178        resolution: &CandleResolution,
179        batched: bool,
180    ) -> Message {
181        let resolution_str = serde_json::to_string(resolution).unwrap_or_default();
182        let resolution_str = resolution_str.trim_matches('"');
183        MessageFormatter::sub_message(
184            CHANNEL,
185            json!({"id": format!("{id}/{resolution_str}"), "batched": batched}),
186        )
187    }
188
189    pub(crate) fn unsub_message(id: &Ticker, resolution: &CandleResolution) -> Message {
190        let resolution_str = serde_json::to_string(resolution).unwrap_or_default();
191        let resolution_str = resolution_str.trim_matches('"');
192        MessageFormatter::unsub_message(CHANNEL, json!({"id": format!("{id}/{resolution_str}")}))
193    }
194}
195
196pub(crate) mod block_height {
197    use super::{json, Message, MessageFormatter};
198    pub const CHANNEL: &str = "v4_block_height";
199
200    pub(crate) fn sub_message(batched: bool) -> Message {
201        MessageFormatter::sub_message(CHANNEL, json!({"batched": batched}))
202    }
203
204    pub(crate) fn unsub_message() -> Message {
205        MessageFormatter::unsub_message(CHANNEL, json!({}))
206    }
207}
208
209/* Main WS type */
210#[allow(clippy::large_enum_variant)]
211#[derive(Debug, Deserialize)]
212#[serde(tag = "type")]
213pub(crate) enum WsMessage {
214    #[serde(rename = "connected")]
215    Setup(StatusConnectedMessage),
216    #[serde(rename = "error")]
217    Error(StatusErrorMessage),
218    #[serde(rename = "unsubscribed")]
219    Unsub(StatusUnsubMessage),
220    #[serde(untagged)]
221    Data(FeedMessage),
222}
223
224#[derive(Debug, Deserialize)]
225pub(crate) struct StatusConnectedMessage {
226    pub(crate) connection_id: String,
227    #[allow(dead_code)] // TODO remove after completion.
228    pub(crate) message_id: u64,
229}
230
231#[derive(Debug, Deserialize)]
232pub(crate) struct StatusErrorMessage {
233    pub(crate) message: String,
234    #[allow(dead_code)] // TODO remove after completion.
235    pub(crate) connection_id: String,
236    #[allow(dead_code)] // TODO remove after completion.
237    pub(crate) message_id: u64,
238}
239
240#[derive(Debug, Deserialize)]
241pub(crate) struct StatusUnsubMessage {
242    #[allow(dead_code)] // TODO remove after completion.
243    pub(crate) connection_id: String,
244    #[allow(dead_code)] // TODO remove after completion.
245    pub(crate) message_id: u64,
246    pub(crate) channel: String,
247    pub(crate) id: Option<String>,
248}
249
250/// Feed Types
251#[derive(Debug, Deserialize)]
252#[serde(tag = "channel")]
253pub enum FeedMessage {
254    /// Subaccounts.
255    #[serde(rename = "v4_subaccounts")]
256    Subaccounts(SubaccountsMessage),
257    /// Orders.
258    #[serde(rename = "v4_orderbook")]
259    Orders(OrdersMessage),
260    /// Trades.
261    #[serde(rename = "v4_trades")]
262    Trades(TradesMessage),
263    /// Markets.
264    #[serde(rename = "v4_markets")]
265    Markets(MarketsMessage),
266    /// Candles.
267    #[serde(rename = "v4_candles")]
268    Candles(CandlesMessage),
269    /// Parent subaccounts.
270    #[serde(rename = "v4_parent_subaccounts")]
271    ParentSubaccounts(ParentSubaccountsMessage),
272    /// Block height.
273    #[serde(rename = "v4_block_height")]
274    BlockHeight(BlockHeightMessage),
275}
276
277macro_rules! impl_feed_message_try_from {
278    ($target_type:ty, $variant:ident) => {
279        impl TryFrom<FeedMessage> for $target_type {
280            type Error = ();
281            fn try_from(value: FeedMessage) -> Result<Self, Self::Error> {
282                match value {
283                    FeedMessage::$variant(a) => Ok(a),
284                    _ => Err(()),
285                }
286            }
287        }
288    };
289}
290
291/// Subaccounts message.
292#[derive(Debug, Deserialize)]
293#[serde(tag = "type")]
294pub enum SubaccountsMessage {
295    /// Initial.
296    #[serde(rename = "subscribed")]
297    Initial(SubaccountsInitialMessage),
298    /// Update.
299    #[serde(untagged)]
300    Update(SubaccountsUpdateMessage),
301}
302impl_feed_message_try_from!(SubaccountsMessage, Subaccounts);
303
304/// Subaccounts message.
305#[derive(Debug, Deserialize)]
306#[serde(tag = "type")]
307pub enum ParentSubaccountsMessage {
308    /// Initial.
309    #[serde(rename = "subscribed")]
310    Initial(ParentSubaccountsInitialMessage),
311    /// Update.
312    #[serde(untagged)]
313    Update(ParentSubaccountsUpdateMessage),
314}
315impl_feed_message_try_from!(ParentSubaccountsMessage, ParentSubaccounts);
316
317/// Trades message.
318#[derive(Debug, Deserialize)]
319#[serde(tag = "type")]
320pub enum TradesMessage {
321    /// Initial.
322    #[serde(rename = "subscribed")]
323    Initial(TradesInitialMessage),
324    /// Update.
325    #[serde(untagged)]
326    Update(TradesUpdateMessage),
327}
328impl_feed_message_try_from!(TradesMessage, Trades);
329
330/// Orders message.
331#[derive(Debug, Deserialize)]
332#[serde(tag = "type")]
333pub enum OrdersMessage {
334    /// Initial.
335    #[serde(rename = "subscribed")]
336    Initial(OrdersInitialMessage),
337    /// Update.
338    #[serde(untagged)]
339    Update(OrdersUpdateMessage),
340}
341impl_feed_message_try_from!(OrdersMessage, Orders);
342
343/// Markets message.
344#[derive(Debug, Deserialize)]
345#[serde(tag = "type")]
346pub enum MarketsMessage {
347    /// Initial.
348    #[serde(rename = "subscribed")]
349    Initial(MarketsInitialMessage),
350    /// Update.
351    #[serde(untagged)]
352    Update(MarketsUpdateMessage),
353}
354impl_feed_message_try_from!(MarketsMessage, Markets);
355
356/// Candles message.
357#[derive(Debug, Deserialize)]
358#[serde(tag = "type")]
359pub enum CandlesMessage {
360    /// Initial.
361    #[serde(rename = "subscribed")]
362    Initial(CandlesInitialMessage),
363    /// Update.
364    #[serde(untagged)]
365    Update(CandlesUpdateMessage),
366}
367impl_feed_message_try_from!(CandlesMessage, Candles);
368
369/// Block height message.
370#[derive(Debug, Deserialize)]
371#[serde(tag = "type")]
372pub enum BlockHeightMessage {
373    /// Initial.
374    #[serde(rename = "subscribed")]
375    Initial(BlockHeightInitialMessage),
376    /// Update.
377    #[serde(untagged)]
378    Update(BlockHeightUpdateMessage),
379}
380impl_feed_message_try_from!(BlockHeightMessage, BlockHeight);
381
382impl FeedMessage {
383    pub(crate) fn subscription(&self) -> Option<Subscription> {
384        let parse_subacc_id = |id: &str| -> Option<Subaccount> {
385            // Parse "id": "Address/Number"
386            let mut id_split = id.split('/');
387            let address = id_split.next()?.parse().ok()?;
388            let number_str = id_split.next()?;
389            let number = serde_json::from_str::<SubaccountNumber>(number_str).ok()?;
390            Some(Subaccount::new(address, number))
391        };
392        let parse_psubacc_id = |id: &str| -> Option<ParentSubaccount> {
393            // Parse "id": "Address/Number"
394            let mut id_split = id.split('/');
395            let address = id_split.next()?.parse().ok()?;
396            let number_str = id_split.next()?;
397            let number = serde_json::from_str::<ParentSubaccountNumber>(number_str).ok()?;
398            Some(ParentSubaccount::new(address, number))
399        };
400        let parse_candles_id = |id: &str| -> Option<(Ticker, CandleResolution)> {
401            // Parse "id": "TICKER/RESOLUTION"
402            let mut id_split = id.split('/');
403            let ticker = Ticker(id_split.next()?.into());
404            let resolution_str = format!("\"{}\"", id_split.next()?);
405            let resolution = serde_json::from_str(&resolution_str).ok()?;
406            Some((ticker, resolution))
407        };
408
409        match self {
410            Self::Subaccounts(SubaccountsMessage::Initial(msg)) => {
411                let subacc = parse_subacc_id(&msg.id)?;
412                Some(Subscription::Subaccounts(subacc))
413            }
414            Self::Subaccounts(SubaccountsMessage::Update(msg)) => {
415                let subacc = parse_subacc_id(&msg.id)?;
416                Some(Subscription::Subaccounts(subacc))
417            }
418
419            Self::ParentSubaccounts(ParentSubaccountsMessage::Initial(msg)) => {
420                let subacc = parse_psubacc_id(&msg.id)?;
421                Some(Subscription::ParentSubaccounts(subacc))
422            }
423            Self::ParentSubaccounts(ParentSubaccountsMessage::Update(msg)) => {
424                let subacc = parse_psubacc_id(&msg.id)?;
425                Some(Subscription::ParentSubaccounts(subacc))
426            }
427
428            Self::Orders(OrdersMessage::Initial(msg)) => {
429                Some(Subscription::Orders(Ticker(msg.id.clone())))
430            }
431            Self::Orders(OrdersMessage::Update(msg)) => {
432                Some(Subscription::Orders(Ticker(msg.id.clone())))
433            }
434
435            Self::Trades(TradesMessage::Initial(msg)) => {
436                Some(Subscription::Trades(Ticker(msg.id.clone())))
437            }
438            Self::Trades(TradesMessage::Update(msg)) => {
439                Some(Subscription::Trades(Ticker(msg.id.clone())))
440            }
441
442            Self::Markets(MarketsMessage::Update(_)) => Some(Subscription::Markets),
443            Self::Markets(MarketsMessage::Initial(_)) => Some(Subscription::Markets),
444
445            Self::Candles(CandlesMessage::Initial(msg)) => {
446                let (ticker, resolution) = parse_candles_id(&msg.id)?;
447                Some(Subscription::Candles(ticker, resolution))
448            }
449            Self::Candles(CandlesMessage::Update(msg)) => {
450                let (ticker, resolution) = parse_candles_id(&msg.id)?;
451                Some(Subscription::Candles(ticker, resolution))
452            }
453
454            Self::BlockHeight(BlockHeightMessage::Initial(_)) => Some(Subscription::BlockHeight),
455            Self::BlockHeight(BlockHeightMessage::Update(_)) => Some(Subscription::BlockHeight),
456        }
457    }
458}
459
460/// Subaccount initial.
461#[derive(Debug, Deserialize)]
462pub struct SubaccountsInitialMessage {
463    /// Connection id.
464    pub connection_id: String,
465    /// Subaccount.
466    pub contents: SubaccountsInitialMessageContents,
467    /// Id.
468    pub id: String,
469    /// Message id.
470    pub message_id: u64,
471}
472
473/// Subaccount.
474#[derive(Debug, Deserialize)]
475#[serde(rename_all = "camelCase")]
476#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
477pub struct SubaccountsInitialMessageContents {
478    /// Subaccount.
479    pub subaccount: SubaccountMessageObject,
480    /// Orders.
481    pub orders: Vec<OrderMessageObject>,
482    /// Block height.
483    pub block_height: Height,
484}
485
486/// Parent subaccount initial.
487#[derive(Debug, Deserialize)]
488pub struct ParentSubaccountsInitialMessage {
489    /// Connection id.
490    pub connection_id: String,
491    /// Subaccount.
492    pub contents: ParentSubaccountsInitialMessageContents,
493    /// Id.
494    pub id: String,
495    /// Message id.
496    pub message_id: u64,
497}
498
499/// Parent subaccount.
500#[derive(Debug, Deserialize)]
501#[serde(rename_all = "camelCase")]
502#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
503pub struct ParentSubaccountsInitialMessageContents {
504    /// Subaccount.
505    pub subaccount: ParentSubaccountMessageObject,
506    /// Orders.
507    pub orders: Vec<OrderMessageObject>,
508    /// Block height.
509    pub block_height: Height,
510}
511
512/// Orders initial message.
513#[derive(Debug, Deserialize)]
514pub struct OrdersInitialMessage {
515    /// Connection id.
516    pub connection_id: String,
517    /// Orders.
518    pub contents: OrdersInitialMessageContents,
519    /// Id.
520    pub id: String,
521    /// Message id.
522    pub message_id: u64,
523}
524
525/// Trades initial message.
526#[derive(Debug, Deserialize)]
527pub struct TradesInitialMessage {
528    /// Connection id.
529    pub connection_id: String,
530    /// Trades.
531    pub contents: TradesInitialMessageContents,
532    /// Id.
533    pub id: String,
534    /// Message id.
535    pub message_id: u64,
536}
537
538/// Markets initial message.
539#[derive(Debug, Deserialize)]
540pub struct MarketsInitialMessage {
541    /// Connection id.
542    pub connection_id: String,
543    /// Market.
544    pub contents: MarketsInitialMessageContents,
545    /// Message id.
546    pub message_id: u64,
547}
548
549/// Candles initial message.
550#[derive(Debug, Deserialize)]
551pub struct CandlesInitialMessage {
552    /// Connection id.
553    pub connection_id: String,
554    /// Candles.
555    pub contents: CandlesInitialMessageContents,
556    /// Id.
557    pub id: String,
558    /// Message id.
559    pub message_id: u64,
560}
561
562/// Block height initial message.
563#[derive(Debug, Deserialize)]
564pub struct BlockHeightInitialMessage {
565    /// Connection id.
566    pub connection_id: String,
567    /// Block height contents.
568    pub contents: BlockHeightInitialMessageContents,
569    /// Message id.
570    pub message_id: u64,
571}
572
573// Updates
574macro_rules! generate_contents_deserialize_function {
575    ($fn_name:ident, $result_type:ty) => {
576        fn $fn_name<'de, D>(deserializer: D) -> Result<Vec<$result_type>, D::Error>
577        where
578            D: serde::Deserializer<'de>,
579        {
580            let value = Value::deserialize(deserializer)?;
581
582            match value {
583                // Batched
584                Value::Array(arr) => arr
585                    .into_iter()
586                    .map(|v| serde_json::from_value(v))
587                    .collect::<Result<Vec<$result_type>, _>>()
588                    .map_err(serde::de::Error::custom),
589                // Streamed
590                Value::Object(obj) => {
591                    let item = serde_json::from_value::<$result_type>(Value::Object(obj.clone()))
592                        .map_err(serde::de::Error::custom)?;
593                    Ok(vec![item])
594                }
595                _ => Err(serde::de::Error::custom("Expected array or object")),
596            }
597        }
598    };
599}
600
601/// Subaccount update.
602#[derive(Debug, Deserialize)]
603pub struct SubaccountsUpdateMessage {
604    /// Connection id.
605    pub connection_id: String,
606    /// Update.
607    #[serde(deserialize_with = "deserialize_subaccounts_contents")]
608    pub contents: Vec<SubaccountUpdateMessageContents>,
609    /// Id.
610    pub id: String,
611    /// Message id.
612    pub message_id: u64,
613    /// Version.
614    pub version: String,
615}
616generate_contents_deserialize_function!(
617    deserialize_subaccounts_contents,
618    SubaccountUpdateMessageContents
619);
620
621/// Subaccount update contents.
622#[derive(Clone, Debug, Deserialize)]
623#[serde(rename_all = "camelCase")]
624#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
625pub struct SubaccountUpdateMessageContents {
626    /// Perpetual position updates on the subaccount.
627    pub perpetual_positions: Option<Vec<PerpetualPositionSubaccountMessageContents>>,
628    /// Asset position updates on the subaccount.
629    pub asset_positions: Option<Vec<AssetPositionSubaccountMessageContents>>,
630    /// Order updates on the subaccount.
631    pub orders: Option<Vec<OrderSubaccountMessageContents>>,
632    /// Fills that occur on the subaccount.
633    pub fills: Option<Vec<FillSubaccountMessageContents>>,
634    /// Transfers that occur on the subaccount.
635    pub transfers: Option<TransferSubaccountMessageContents>,
636    /// Rewards that occur on the subaccount.
637    pub trading_reward: Option<TradingRewardSubaccountMessageContents>,
638    /// Block height.
639    pub block_height: Option<Height>,
640}
641
642/// Perpetual position on subaccount.
643#[derive(Clone, Debug, Deserialize)]
644#[serde(rename_all = "camelCase")]
645#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
646pub struct PerpetualPositionSubaccountMessageContents {
647    /// Address.
648    pub address: Address,
649    /// Subaccount number.
650    pub subaccount_number: SubaccountNumber,
651    /// Position id.
652    pub position_id: String,
653    /// Market ticker.
654    pub market: Ticker,
655    /// Side (buy/sell).
656    pub side: PositionSide,
657    /// Position status.
658    pub status: PerpetualPositionStatus,
659    /// Size.
660    pub size: Quantity,
661    /// Maximum size.
662    pub max_size: Quantity,
663    /// Net funding.
664    pub net_funding: BigDecimal,
665    /// Entry price.
666    pub entry_price: Price,
667    /// Exit price.
668    pub exit_price: Option<Price>,
669    /// Sum at open.
670    pub sum_open: BigDecimal,
671    /// Sum at close.
672    pub sum_close: BigDecimal,
673    /// Actual PnL.
674    pub realized_pnl: Option<BigDecimal>,
675    /// Potential PnL.
676    pub unrealized_pnl: Option<BigDecimal>,
677}
678
679/// Asset position per subaccount.
680#[derive(Clone, Debug, Deserialize)]
681#[serde(rename_all = "camelCase")]
682#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
683pub struct AssetPositionSubaccountMessageContents {
684    /// Address.
685    pub address: Address,
686    /// Subaccount number.
687    pub subaccount_number: SubaccountNumber,
688    /// Position id.
689    pub position_id: String,
690    /// Asset id.
691    pub asset_id: AssetId,
692    /// Token symbol.
693    pub symbol: Symbol,
694    /// Side (buy/sell).
695    pub side: PositionSide,
696    /// Size.
697    pub size: Quantity,
698}
699
700/// Order per subaccount.
701#[derive(Clone, Debug, Deserialize)]
702#[serde(rename_all = "camelCase")]
703#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
704pub struct OrderSubaccountMessageContents {
705    /// Id.
706    pub id: String,
707    /// Subaccount id.
708    pub subaccount_id: SubaccountId,
709    /// Client id.
710    pub client_id: ClientId,
711    /// Clob pair id.
712    pub clob_pair_id: Option<ClobPairId>,
713    /// Side (buy/sell).
714    pub side: Option<OrderSide>,
715    /// Size.
716    pub size: Option<Quantity>,
717    /// Market ticker.
718    pub ticker: Option<Ticker>,
719    /// Price.
720    pub price: Option<Price>,
721    #[serde(rename = "type")]
722    /// Order type.
723    pub order_type: Option<OrderType>,
724    /// Time-in-force.
725    pub time_in_force: Option<ApiTimeInForce>,
726    /// Post-only.
727    pub post_only: Option<bool>,
728    /// Reduce-only.
729    pub reduce_only: Option<bool>,
730    /// Order status.
731    pub status: ApiOrderStatus,
732    /// Order flags.
733    pub order_flags: OrderFlags,
734    /// Total filled.
735    pub total_filled: Option<BigDecimal>,
736    /// Total optimistic filled.
737    pub total_optimistic_filled: Option<BigDecimal>,
738    /// Block height.
739    pub good_til_block: Option<Height>,
740    /// Time(UTC).
741    pub good_til_block_time: Option<DateTime<Utc>>,
742    /// Trigger price.
743    pub trigger_price: Option<Price>,
744    /// Time(UTC).
745    pub updated_at: Option<DateTime<Utc>>,
746    /// Block height.
747    pub updated_at_height: Option<Height>,
748    /// Removal reason.
749    pub removal_reason: Option<String>,
750    /// Block height.
751    pub created_at_height: Option<Height>,
752    /// Client metadata.
753    pub client_metadata: Option<ClientMetadata>,
754}
755
756/// Fill per subaccount.
757#[derive(Clone, Debug, Deserialize)]
758#[serde(rename_all = "camelCase")]
759#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
760pub struct FillSubaccountMessageContents {
761    /// Fill id.
762    pub id: FillId,
763    /// Subaccount id.
764    pub subaccount_id: SubaccountId,
765    /// Order side.
766    pub side: OrderSide,
767    /// Liquidity.
768    pub liquidity: Liquidity,
769    /// Fill type.
770    #[serde(rename = "type")]
771    pub fill_type: FillType,
772    /// Clob pair id.
773    pub clob_pair_id: ClobPairId,
774    /// Size.
775    pub size: Quantity,
776    /// Price.
777    pub price: Price,
778    /// Quote amount.
779    pub quote_amount: String,
780    /// Event id.
781    pub event_id: String,
782    /// Transaction hash.
783    pub transaction_hash: String,
784    /// Time(UTC).
785    pub created_at: DateTime<Utc>,
786    /// Block height.
787    pub created_at_height: Height,
788    /// Market ticker.
789    pub ticker: Ticker,
790    /// Order id.
791    pub order_id: Option<OrderId>,
792    /// Client metadata.
793    pub client_metadata: Option<ClientMetadata>,
794}
795
796/// Transfer per subaccount.
797#[derive(Clone, Debug, Deserialize)]
798#[serde(rename_all = "camelCase")]
799#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
800pub struct TransferSubaccountMessageContents {
801    /// Sender.
802    pub sender: Account,
803    /// Recipient.
804    pub recipient: Account,
805    /// Token symbol.
806    pub symbol: Symbol,
807    /// Size.
808    pub size: Quantity,
809    /// Transfer type.
810    #[serde(rename = "type")]
811    pub transfer_type: TransferType,
812    /// Transaction hash.
813    pub transaction_hash: String,
814    /// Time(UTC).
815    pub created_at: DateTime<Utc>,
816    /// Block height.
817    pub created_at_height: Height,
818}
819
820/// Trading reward.
821#[derive(Clone, Debug, Deserialize)]
822#[serde(rename_all = "camelCase")]
823#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
824pub struct TradingRewardSubaccountMessageContents {
825    /// Trading reward.
826    pub trading_reward: BigDecimal,
827    /// Time(UTC).
828    pub created_at: DateTime<Utc>,
829    /// Block height.
830    pub created_at_height: Height,
831}
832
833/// Subaccount update.
834#[derive(Debug, Deserialize)]
835pub struct ParentSubaccountsUpdateMessage {
836    /// Connection id.
837    pub connection_id: String,
838    /// Update.
839    #[serde(deserialize_with = "deserialize_subaccounts_contents")]
840    pub contents: Vec<SubaccountUpdateMessageContents>,
841    /// Id.
842    pub id: String,
843    /// Message id.
844    pub message_id: u64,
845    /// Version.
846    pub version: String,
847}
848
849/// Subaccount update contents.
850#[derive(Clone, Debug, Deserialize)]
851#[serde(rename_all = "camelCase")]
852#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
853pub struct ParentSubaccountUpdateMessageContents {
854    /// Perpetual position updates on the subaccount.
855    pub perpetual_positions: Option<Vec<PerpetualPositionSubaccountMessageContents>>,
856    /// Asset position updates on the subaccount.
857    pub asset_positions: Option<Vec<AssetPositionSubaccountMessageContents>>,
858    /// Order updates on the subaccount.
859    pub orders: Option<Vec<OrderSubaccountMessageContents>>,
860    /// Fills that occur on the subaccount.
861    pub fills: Option<Vec<FillSubaccountMessageContents>>,
862    /// Transfers that occur on the subaccount.
863    pub transfers: Option<TransferSubaccountMessageContents>,
864    /// Rewards that occur on the subaccount.
865    pub trading_reward: Option<TradingRewardSubaccountMessageContents>,
866    /// Block height.
867    pub block_height: Option<Height>,
868}
869
870/// Order update message.
871#[derive(Debug, Deserialize)]
872pub struct OrdersUpdateMessage {
873    /// Connection id.
874    pub connection_id: String,
875    /// Update.
876    #[serde(deserialize_with = "deserialize_orders_contents")]
877    pub contents: OrdersUpdateMessageContents,
878    /// Id.
879    pub id: String,
880    /// Message id.
881    pub message_id: u64,
882    /// Version.
883    pub version: String,
884}
885
886fn deserialize_orders_contents<'de, D>(
887    deserializer: D,
888) -> Result<OrdersUpdateMessageContents, D::Error>
889where
890    D: serde::Deserializer<'de>,
891{
892    let value = Value::deserialize(deserializer)?;
893
894    match value {
895        // Batched
896        Value::Array(arr) => {
897            let mut bids = Vec::new();
898            let mut asks = Vec::new();
899
900            for v in arr {
901                let item: OrdersUpdateMessageContents =
902                    serde_json::from_value(v).map_err(serde::de::Error::custom)?;
903
904                if let Some(item_bids) = item.bids {
905                    bids.extend(item_bids);
906                }
907                if let Some(item_asks) = item.asks {
908                    asks.extend(item_asks);
909                }
910            }
911
912            Ok(OrdersUpdateMessageContents {
913                bids: if bids.is_empty() { None } else { Some(bids) },
914                asks: if asks.is_empty() { None } else { Some(asks) },
915            })
916        }
917        // Streamed
918        Value::Object(obj) => {
919            let item =
920                serde_json::from_value::<OrdersUpdateMessageContents>(Value::Object(obj.clone()))
921                    .map_err(serde::de::Error::custom)?;
922            Ok(item)
923        }
924        _ => Err(serde::de::Error::custom("Expected array or object")),
925    }
926}
927
928/// Orderbook update.
929#[derive(Deserialize, Debug, Clone)]
930pub struct OrdersUpdateMessageContents {
931    /// Bids.
932    pub bids: Option<Vec<OrderbookResponsePriceLevel>>,
933    /// Asks.
934    pub asks: Option<Vec<OrderbookResponsePriceLevel>>,
935}
936
937/// Trades update.
938#[derive(Deserialize, Debug, Clone)]
939pub struct TradesUpdateMessage {
940    /// Connection id.
941    pub connection_id: String,
942    /// Update.
943    #[serde(deserialize_with = "deserialize_trades_contents")]
944    pub contents: Vec<TradesUpdateMessageContents>,
945    /// Id.
946    pub id: String,
947    /// Message id.
948    pub message_id: u64,
949    /// Version.
950    pub version: String,
951}
952generate_contents_deserialize_function!(deserialize_trades_contents, TradesUpdateMessageContents);
953
954/// Trades updates.
955#[derive(Deserialize, Debug, Clone)]
956#[serde(rename_all = "camelCase")]
957#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
958pub struct TradesUpdateMessageContents {
959    /// Updates.
960    pub trades: Vec<TradeUpdate>,
961}
962
963/// Trade update.
964#[derive(Deserialize, Debug, Clone)]
965#[serde(rename_all = "camelCase")]
966#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
967pub struct TradeUpdate {
968    /// Unique id of the trade, which is the taker fill id.
969    pub id: TradeId,
970    /// Time(UTC).
971    pub created_at: DateTime<Utc>,
972    /// Side (buy/sell).
973    pub side: OrderSide,
974    /// Price.
975    pub price: Price,
976    /// Size.
977    pub size: Quantity,
978    /// Trade type.
979    #[serde(rename = "type")]
980    pub trade_type: TradeType,
981}
982
983/// Markets update message.
984#[derive(Debug, Deserialize)]
985pub struct MarketsUpdateMessage {
986    /// Connection id.
987    pub connection_id: String,
988    /// Updates.
989    #[serde(deserialize_with = "deserialize_markets_contents")]
990    pub contents: Vec<MarketsUpdateMessageContents>,
991    /// Message id.
992    pub message_id: u64,
993    /// Version.
994    pub version: String,
995}
996generate_contents_deserialize_function!(deserialize_markets_contents, MarketsUpdateMessageContents);
997
998/// Markets update.
999#[derive(Debug, Deserialize)]
1000#[serde(rename_all = "camelCase")]
1001#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
1002pub struct MarketsUpdateMessageContents {
1003    /// Trading.
1004    pub trading: Option<HashMap<Ticker, TradingPerpetualMarket>>,
1005    /// Oracle prices.
1006    pub oracle_prices: Option<HashMap<Ticker, OraclePriceMarket>>,
1007}
1008
1009/// Perpetual market info.
1010#[derive(Deserialize, Debug, Clone)]
1011#[serde(rename_all = "camelCase")]
1012#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
1013pub struct TradingPerpetualMarket {
1014    /// Atomic resolution
1015    pub atomic_resolution: Option<i32>,
1016    /// Base asset.
1017    pub base_asset: Option<String>,
1018    /// Base open interest.
1019    pub base_open_interest: Option<BigDecimal>,
1020    /// Base position size.
1021    pub base_position_size: Option<Quantity>,
1022    /// Clob pair id.
1023    pub clob_pair_id: Option<ClobPairId>,
1024    /// Id.
1025    pub id: Option<String>,
1026    /// Market id.
1027    pub market_id: Option<u64>,
1028    /// Incremental position size.
1029    pub incremental_position_size: Option<Quantity>,
1030    /// Initial margin fraction.
1031    pub initial_margin_fraction: Option<BigDecimal>,
1032    /// Maintenance margin fraction.
1033    pub maintenance_margin_fraction: Option<BigDecimal>,
1034    /// Max position size.
1035    pub max_position_size: Option<Quantity>,
1036    /// Open interest.
1037    pub open_interest: Option<BigDecimal>,
1038    /// Quantum conversion exponent.
1039    pub quantum_conversion_exponent: Option<i32>,
1040    /// Quote asset.
1041    pub quote_asset: Option<String>,
1042    /// Market status
1043    pub status: Option<PerpetualMarketStatus>,
1044    /// Step base quantums.
1045    pub step_base_quantums: Option<i32>,
1046    /// Subticks per tick.
1047    pub subticks_per_tick: Option<i32>,
1048    /// Market ticker.
1049    pub ticker: Option<Ticker>,
1050    /// 24-h price change.
1051    #[serde(rename = "priceChange24H")]
1052    pub price_change_24h: Option<BigDecimal>,
1053    /// 24-h number of trades.
1054    #[serde(rename = "trades24H")]
1055    pub trades_24h: Option<u64>,
1056    /// 24-h volume.
1057    #[serde(rename = "volume24H")]
1058    pub volume_24h: Option<Quantity>,
1059    /// Next funding rate.
1060    pub next_funding_rate: Option<BigDecimal>,
1061}
1062
1063/// Oracle price for market.
1064#[derive(Debug, Deserialize)]
1065#[serde(rename_all = "camelCase")]
1066#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
1067pub struct OraclePriceMarket {
1068    /// Oracle price.
1069    pub oracle_price: Price,
1070    /// Time(UTC).
1071    pub effective_at: DateTime<Utc>,
1072    /// Block height.
1073    pub effective_at_height: Height,
1074    /// Market id.
1075    pub market_id: u64,
1076}
1077
1078/// Candles update.
1079#[derive(Debug, Deserialize)]
1080pub struct CandlesUpdateMessage {
1081    /// Connection id.
1082    pub connection_id: String,
1083    /// Candle.
1084    #[serde(deserialize_with = "deserialize_candles_contents")]
1085    pub contents: Vec<Candle>,
1086    /// Id.
1087    pub id: String,
1088    /// Message id.
1089    pub message_id: u64,
1090    /// Version.
1091    pub version: String,
1092}
1093generate_contents_deserialize_function!(deserialize_candles_contents, Candle);
1094
1095/// Block height update message.
1096#[derive(Debug, Deserialize)]
1097pub struct BlockHeightUpdateMessage {
1098    /// Connection id.
1099    pub connection_id: String,
1100    /// Updates.
1101    #[serde(deserialize_with = "deserialize_block_height_contents")]
1102    pub contents: Vec<BlockHeightUpdateMessageContents>,
1103    /// Message id.
1104    pub message_id: u64,
1105    /// Version.
1106    pub version: String,
1107}
1108generate_contents_deserialize_function!(
1109    deserialize_block_height_contents,
1110    BlockHeightUpdateMessageContents
1111);
1112
1113/// Block height update message contents.
1114#[derive(Debug, Deserialize)]
1115#[serde(rename_all = "camelCase")]
1116#[cfg_attr(any(test, feature = "strict-serde"), serde(deny_unknown_fields))]
1117pub struct BlockHeightUpdateMessageContents {
1118    /// Block height.
1119    pub block_height: Height,
1120    /// Time of block.
1121    pub time: DateTime<Utc>,
1122}
dydx/indexer/sock/messages.rs

dydx/indexer/sock/
messages.rs
