df_hyperliquid_core/live/
frame.rs

1//! Received frames (`{"channel":..,"data":..}`) read into a type per channel ([`decode`]), numbers
2//! as exact decimals, never floats. An unknown channel is [`Frame::Other`], kept as written; a
3//! frame that does not read is a [`BadFrame`], never a panic.
4
5use std::{borrow::Cow, fmt};
6
7use crate::client::ExchangeResponse;
8use crate::live::account::{
9    AllDexsState, AssetData, DexState, HistoricalOrders, Ledger, OpenOrders, OrderUpdate, SpotState, TwapHistory,
10    TwapSliceFills, TwapStates, UserEvent, UserFills, UserFundings, WebData3,
11};
12use crate::live::market::{Bbo, Book, Candle, Mids, PerpCtx, SpotCtx, Trade};
13use crate::live::subscription::{BadSubscription, Subscription};
14use crate::utils::Json;
15use serde::Deserialize;
16use serde_json::value::RawValue;
17
18/// One frame, by channel.
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub enum Frame {
21    /// `pong`: the answer to a ping.
22    Pong,
23    /// `subscriptionResponse`: the venue took a subscribe or an unsubscribe.
24    Ack(Ack),
25    /// `error`: the venue refused something sent, in its own words.
26    Error(Refusal),
27    /// `post`: the answer to a post, by its id.
28    Post(Reply),
29    AllMids(Mids),
30    /// `l2Book`.
31    Book(Book),
32    Trades(Vec<Trade>),
33    Bbo(Bbo),
34    Candle(Candle),
35    /// `activeAssetCtx`: a perp's context.
36    AssetCtx {
37        coin: String,
38        ctx: PerpCtx,
39    },
40    /// `activeSpotAssetCtx`: what `activeAssetCtx` sends for a spot pair or an outcome side.
41    SpotAssetCtx {
42        coin: String,
43        ctx: SpotCtx,
44    },
45    /// `activeAssetData`.
46    AssetData(AssetData),
47    UserFills(UserFills),
48    OrderUpdates(Vec<OrderUpdate>),
49    /// `user`: what `userEvents` sends.
50    UserEvent(UserEvent),
51    UserFundings(UserFundings),
52    /// `userNonFundingLedgerUpdates`.
53    Ledger(Ledger),
54    ClearinghouseState(DexState),
55    AllDexsClearinghouseState(AllDexsState),
56    OpenOrders(OpenOrders),
57    SpotState(SpotState),
58    TwapStates(TwapStates),
59    /// `userTwapSliceFills`.
60    TwapSliceFills(TwapSliceFills),
61    /// `userTwapHistory`.
62    TwapHistory(TwapHistory),
63    Notification(String),
64    WebData3(WebData3),
65    /// Every perp dex's contexts, by dex name, the first `""`.
66    AllDexsAssetCtxs(Vec<(String, Vec<PerpCtx>)>),
67    AssetCtxs {
68        dex: String,
69        ctxs: Vec<PerpCtx>,
70    },
71    SpotAssetCtxs(Vec<SpotCtx>),
72    /// What changed in `outcomeMeta`, kept as sent: `[{"outcomeSettled":23205}]`. Its arrival
73    /// is the news; read `outcomeMeta` again for the whole.
74    OutcomeMetaUpdates(Json),
75    /// `fastAssetCtxs`: base64 of raw DEFLATE (RFC 1951) over JSON, kept as sent. Inflating it
76    /// needs a decompressor this crate does not carry.
77    FastAssetCtxs(String),
78    /// `userHistoricalOrders`.
79    HistoricalOrders(HistoricalOrders),
80    /// A channel this build does not read: its name, and its data as written (empty when the
81    /// frame has none).
82    Other {
83        channel: String,
84        data: String,
85    },
86}
87
88/// Which way a subscription went.
89#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
90#[serde(rename_all = "lowercase")]
91pub enum Method {
92    Subscribe,
93    Unsubscribe,
94}
95
96/// The venue's acknowledgement, echoing the subscription as it read it: keys reordered, left
97/// out options filled in, the user in lowercase.
98#[derive(Debug, Clone, PartialEq, Eq)]
99pub struct Ack {
100    pub method: Method,
101    pub echo: Json,
102}
103
104impl Ack {
105    /// The subscription it names.
106    pub fn subscription(&self) -> Result<Subscription, BadSubscription> {
107        Subscription::parse(&self.echo)
108    }
109}
110
111/// The venue refusing something: `Already subscribed: {...}`, `Already unsubscribed: {...}`,
112/// `Error parsing JSON into valid websocket request: {...}` with what was sent.
113#[derive(Debug, Clone, PartialEq, Eq)]
114pub struct Refusal {
115    pub message: String,
116}
117
118impl Refusal {
119    /// The post it refuses: from an `id=<n>` at the end of the message, or the `id` of a post
120    /// it quotes back.
121    #[must_use]
122    pub fn post_id(&self) -> Option<u64> {
123        if let Some((_, digits)) = self.message.rsplit_once("id=")
124            && !digits.is_empty()
125            && digits.bytes().all(|byte| byte.is_ascii_digit())
126        {
127            return digits.parse().ok();
128        }
129        let quoted = self.quoted()?;
130        match quoted.get("method").and_then(Json::as_str) {
131            Some("post") => quoted.get("id").and_then(Json::as_u64),
132            _ => None,
133        }
134    }
135
136    /// The subscription it refuses, when it quotes one this crate reads.
137    #[must_use]
138    pub fn subscription(&self) -> Option<Subscription> {
139        let quoted = self.quoted()?;
140        let subscription = match quoted.get("subscription") {
141            Some(inside) => inside,
142            None if ["Already subscribed", "Already unsubscribed", "Invalid subscription"]
143                .iter()
144                .any(|start| self.message.starts_with(start)) =>
145            {
146                &quoted
147            }
148            None => return None,
149        };
150        Subscription::parse(subscription).ok()
151    }
152
153    /// The JSON the message ends with, when it ends with some.
154    fn quoted(&self) -> Option<Json> {
155        let at = self.message.find('{')?;
156        Json::parse(self.message.get(at..)?).ok()
157    }
158}
159
160/// The answer to a post, by the id it was sent with.
161#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct Reply {
163    pub id: u64,
164    pub answer: Answer,
165}
166
167#[derive(Debug, Clone, PartialEq, Eq)]
168pub enum Answer {
169    /// What `/info` would have answered, as written (`null` for an unknown coin's book), for a
170    /// read's own type to deserialize.
171    Info { kind: String, data: String },
172    /// What `/exchange` would have answered.
173    Action(ExchangeResponse),
174    /// The post was not taken: what HTTP would have said.
175    Error(String),
176}
177
178/// A frame that did not read: the channel it named, when it got that far, and why.
179#[derive(Debug, Clone, PartialEq, Eq)]
180pub struct BadFrame {
181    pub channel: Option<String>,
182    pub detail: String,
183}
184
185impl fmt::Display for BadFrame {
186    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
187        match &self.channel {
188            Some(channel) => write!(f, "a `{channel}` frame did not read: {}", self.detail),
189            None => write!(f, "not a frame: {}", self.detail),
190        }
191    }
192}
193
194impl std::error::Error for BadFrame {}
195
196#[derive(Deserialize)]
197struct Head<'a> {
198    #[serde(borrow)]
199    channel: Cow<'a, str>,
200    #[serde(default, borrow)]
201    data: Option<&'a RawValue>,
202}
203
204#[derive(Deserialize)]
205struct RawAck<'a> {
206    method: Method,
207    #[serde(borrow)]
208    subscription: &'a RawValue,
209}
210
211#[derive(Deserialize)]
212struct RawReply<'a> {
213    id: u64,
214    #[serde(borrow)]
215    response: RawResponse<'a>,
216}
217
218#[derive(Deserialize)]
219struct RawResponse<'a> {
220    #[serde(rename = "type", borrow)]
221    kind: Cow<'a, str>,
222    #[serde(borrow)]
223    payload: &'a RawValue,
224}
225
226#[derive(Deserialize)]
227struct RawInfo<'a> {
228    #[serde(rename = "type")]
229    kind: String,
230    #[serde(default, borrow)]
231    data: Option<&'a RawValue>,
232}
233
234#[derive(Deserialize)]
235struct Active<C> {
236    coin: String,
237    ctx: C,
238}
239
240#[derive(Deserialize)]
241struct DexCtxs {
242    dex: String,
243    ctxs: Vec<PerpCtx>,
244}
245
246#[derive(Deserialize)]
247struct EveryDexCtxs {
248    ctxs: Vec<(String, Vec<PerpCtx>)>,
249}
250
251#[derive(Deserialize)]
252struct Note {
253    notification: String,
254}
255
256/// Reads one text frame.
257pub fn decode(text: &str) -> Result<Frame, BadFrame> {
258    let head: Head<'_> = serde_json::from_str(text)
259        .map_err(|error| BadFrame { channel: None, detail: format!("{error} in {}", shown(text)) })?;
260    let channel = head.channel.as_ref();
261    let data = head.data.map(RawValue::get);
262    let typed = Typed { channel, data };
263    Ok(match channel {
264        "pong" => Frame::Pong,
265        "subscriptionResponse" => {
266            let ack: RawAck<'_> = typed.read()?;
267            let echo = Json::parse(ack.subscription.get()).map_err(|error| typed.fail(error.to_string()))?;
268            Frame::Ack(Ack { method: ack.method, echo })
269        }
270        "error" => Frame::Error(Refusal { message: typed.read()? }),
271        "post" => Frame::Post(typed.reply()?),
272        "allMids" => Frame::AllMids(typed.read()?),
273        "l2Book" => Frame::Book(typed.read()?),
274        "trades" => Frame::Trades(typed.read()?),
275        "bbo" => Frame::Bbo(typed.read()?),
276        "candle" => Frame::Candle(typed.read()?),
277        "activeAssetCtx" => {
278            let active: Active<PerpCtx> = typed.read()?;
279            Frame::AssetCtx { coin: active.coin, ctx: active.ctx }
280        }
281        "activeSpotAssetCtx" => {
282            let active: Active<SpotCtx> = typed.read()?;
283            Frame::SpotAssetCtx { coin: active.coin, ctx: active.ctx }
284        }
285        "activeAssetData" => Frame::AssetData(typed.read()?),
286        "userFills" => Frame::UserFills(typed.read()?),
287        "orderUpdates" => Frame::OrderUpdates(typed.read()?),
288        "user" => Frame::UserEvent(typed.read()?),
289        "userFundings" => Frame::UserFundings(typed.read()?),
290        "userNonFundingLedgerUpdates" => Frame::Ledger(typed.read()?),
291        "clearinghouseState" => Frame::ClearinghouseState(typed.read()?),
292        "allDexsClearinghouseState" => Frame::AllDexsClearinghouseState(typed.read()?),
293        "openOrders" => Frame::OpenOrders(typed.read()?),
294        "spotState" => Frame::SpotState(typed.read()?),
295        "twapStates" => Frame::TwapStates(typed.read()?),
296        "userTwapSliceFills" => Frame::TwapSliceFills(typed.read()?),
297        "userTwapHistory" => Frame::TwapHistory(typed.read()?),
298        "notification" => Frame::Notification(typed.read::<Note>()?.notification),
299        "webData3" => Frame::WebData3(typed.read()?),
300        "allDexsAssetCtxs" => Frame::AllDexsAssetCtxs(typed.read::<EveryDexCtxs>()?.ctxs),
301        "assetCtxs" => {
302            let dex: DexCtxs = typed.read()?;
303            Frame::AssetCtxs { dex: dex.dex, ctxs: dex.ctxs }
304        }
305        "spotAssetCtxs" => Frame::SpotAssetCtxs(typed.read()?),
306        "outcomeMetaUpdates" => {
307            Frame::OutcomeMetaUpdates(Json::parse(typed.data()?).map_err(|error| typed.fail(error.to_string()))?)
308        }
309        "fastAssetCtxs" => Frame::FastAssetCtxs(typed.read()?),
310        "userHistoricalOrders" => Frame::HistoricalOrders(typed.read()?),
311        other => Frame::Other { channel: other.to_owned(), data: data.unwrap_or_default().to_owned() },
312    })
313}
314
315/// A frame's data, read for the channel that named it.
316struct Typed<'a> {
317    channel: &'a str,
318    data: Option<&'a str>,
319}
320
321impl<'a> Typed<'a> {
322    fn fail(&self, detail: String) -> BadFrame {
323        BadFrame { channel: Some(self.channel.to_owned()), detail }
324    }
325
326    fn data(&self) -> Result<&'a str, BadFrame> {
327        self.data.ok_or_else(|| self.fail("no data".to_owned()))
328    }
329
330    fn read<T: Deserialize<'a>>(&self) -> Result<T, BadFrame> {
331        let data = self.data()?;
332        serde_json::from_str(data).map_err(|error| self.fail(format!("{error} in {}", shown(data))))
333    }
334
335    fn reply(&self) -> Result<Reply, BadFrame> {
336        let data = self.data()?;
337        let reply: RawReply<'_> =
338            serde_json::from_str(data).map_err(|error| self.fail(format!("{error} in {}", shown(data))))?;
339        let payload = reply.response.payload.get();
340        let answer = match reply.response.kind.as_ref() {
341            "info" => {
342                let info: RawInfo<'_> = serde_json::from_str(payload).map_err(|error| self.fail(error.to_string()))?;
343                Answer::Info { kind: info.kind, data: info.data.map_or("null", RawValue::get).to_owned() }
344            }
345            "action" => Answer::Action(ExchangeResponse::parse(payload).map_err(|error| self.fail(error.to_string()))?),
346            "error" => Answer::Error(serde_json::from_str(payload).map_err(|error| self.fail(error.to_string()))?),
347            other => return Err(self.fail(format!("a post answered as `{other}`"))),
348        };
349        Ok(Reply { id: reply.id, answer })
350    }
351}
352
353/// The start of a text, for an error message.
354fn shown(text: &str) -> String {
355    let start: String = text.chars().take(160).collect();
356    match start.len() < text.len() {
357        true => format!("{start}..."),
358        false => start,
359    }
360}
361
362#[cfg(test)]
363#[path = "tests/frame.rs"]
364mod tests;
df_hyperliquid_core/live/frame.rs

df_hyperliquid_core/live/
frame.rs
