use std::fmt;
use rust_decimal::Decimal;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_json::Value;
use crate::usdm::{
types::{Interval, Level},
ws::{error::UsdmWsError, stream::StreamName},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum SymbolType {
Um,
Cm,
Other(u64),
}
impl Serialize for SymbolType {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_u64(match self {
Self::Um => 1,
Self::Cm => 2,
Self::Other(raw) => *raw,
})
}
}
impl<'de> Deserialize<'de> for SymbolType {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
Ok(match u64::deserialize(deserializer)? {
1 => Self::Um,
2 => Self::Cm,
raw => Self::Other(raw),
})
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct TickerEvent {
#[serde(rename = "e")]
pub event_type: String,
#[serde(rename = "E")]
pub event_time: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "ps")]
pub pair: String,
#[serde(rename = "p", with = "rust_decimal::serde::str")]
pub price_change: Decimal,
#[serde(rename = "P", with = "rust_decimal::serde::str")]
pub price_change_percent: Decimal,
#[serde(rename = "w", with = "rust_decimal::serde::str")]
pub weighted_avg_price: Decimal,
#[serde(rename = "c", with = "rust_decimal::serde::str")]
pub last_price: Decimal,
#[serde(rename = "Q", with = "rust_decimal::serde::str")]
pub last_quantity: Decimal,
#[serde(rename = "o", with = "rust_decimal::serde::str")]
pub open_price: Decimal,
#[serde(rename = "h", with = "rust_decimal::serde::str")]
pub high_price: Decimal,
#[serde(rename = "l", with = "rust_decimal::serde::str")]
pub low_price: Decimal,
#[serde(rename = "v", with = "rust_decimal::serde::str")]
pub volume: Decimal,
#[serde(rename = "q", with = "rust_decimal::serde::str")]
pub quote_volume: Decimal,
#[serde(rename = "O")]
pub open_time: u64,
#[serde(rename = "C")]
pub close_time: u64,
#[serde(rename = "F")]
pub first_trade_id: i64,
#[serde(rename = "L")]
pub last_trade_id: i64,
#[serde(rename = "n")]
pub trade_count: u64,
#[serde(rename = "st")]
pub symbol_type: SymbolType,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct MarkPriceEvent {
#[serde(rename = "e")]
pub event_type: String,
#[serde(rename = "E")]
pub event_time: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "p", with = "rust_decimal::serde::str")]
pub mark_price: Decimal,
#[serde(rename = "ap", with = "rust_decimal::serde::str")]
pub mark_price_moving_average: Decimal,
#[serde(rename = "P", with = "rust_decimal::serde::str")]
pub estimated_settle_price: Decimal,
#[serde(rename = "i", with = "rust_decimal::serde::str")]
pub index_price: Decimal,
#[serde(rename = "r", with = "rust_decimal::serde::str")]
pub funding_rate: Decimal,
#[serde(rename = "T")]
pub next_funding_time: u64,
#[serde(rename = "st")]
pub symbol_type: SymbolType,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct AggTradeEvent {
#[serde(rename = "e")]
pub event_type: String,
#[serde(rename = "E")]
pub event_time: u64,
#[serde(rename = "a")]
pub id: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "p", with = "rust_decimal::serde::str")]
pub price: Decimal,
#[serde(rename = "q", with = "rust_decimal::serde::str")]
pub quantity: Decimal,
#[serde(rename = "nq", with = "rust_decimal::serde::str")]
pub normal_quantity: Decimal,
#[serde(rename = "f")]
pub first_trade_id: u64,
#[serde(rename = "l")]
pub last_trade_id: u64,
#[serde(rename = "T")]
pub trade_time: u64,
#[serde(rename = "m")]
pub is_buyer_maker: bool,
#[serde(rename = "st")]
pub symbol_type: SymbolType,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct KlineEvent {
#[serde(rename = "e")]
pub event_type: String,
#[serde(rename = "E")]
pub event_time: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "k")]
pub kline: KlineBar,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct KlineBar {
#[serde(rename = "t")]
pub open_time: u64,
#[serde(rename = "T")]
pub close_time: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "i")]
pub interval: Interval,
#[serde(rename = "f")]
pub first_trade_id: i64,
#[serde(rename = "L")]
pub last_trade_id: i64,
#[serde(rename = "o", with = "rust_decimal::serde::str")]
pub open: Decimal,
#[serde(rename = "c", with = "rust_decimal::serde::str")]
pub close: Decimal,
#[serde(rename = "h", with = "rust_decimal::serde::str")]
pub high: Decimal,
#[serde(rename = "l", with = "rust_decimal::serde::str")]
pub low: Decimal,
#[serde(rename = "v", with = "rust_decimal::serde::str")]
pub volume: Decimal,
#[serde(rename = "n")]
pub trade_count: u64,
#[serde(rename = "x")]
pub is_closed: bool,
#[serde(rename = "q", with = "rust_decimal::serde::str")]
pub quote_volume: Decimal,
#[serde(rename = "V", with = "rust_decimal::serde::str")]
pub taker_buy_base_volume: Decimal,
#[serde(rename = "Q", with = "rust_decimal::serde::str")]
pub taker_buy_quote_volume: Decimal,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct PartialDepthEvent {
#[serde(rename = "e")]
pub event_type: String,
#[serde(rename = "E")]
pub event_time: u64,
#[serde(rename = "T")]
pub transaction_time: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "ps")]
pub pair: String,
#[serde(rename = "U")]
pub first_update_id: u64,
#[serde(rename = "u")]
pub final_update_id: u64,
#[serde(rename = "pu")]
pub previous_final_update_id: u64,
#[serde(rename = "b")]
pub bids: Vec<Level>,
#[serde(rename = "a")]
pub asks: Vec<Level>,
#[serde(rename = "st")]
pub symbol_type: SymbolType,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct BookTickerEvent {
#[serde(rename = "e")]
pub event_type: String,
#[serde(rename = "u")]
pub update_id: u64,
#[serde(rename = "s")]
pub symbol: String,
#[serde(rename = "ps")]
pub pair: String,
#[serde(rename = "b", with = "rust_decimal::serde::str")]
pub bid_price: Decimal,
#[serde(rename = "B", with = "rust_decimal::serde::str")]
pub bid_quantity: Decimal,
#[serde(rename = "a", with = "rust_decimal::serde::str")]
pub ask_price: Decimal,
#[serde(rename = "A", with = "rust_decimal::serde::str")]
pub ask_quantity: Decimal,
#[serde(rename = "T")]
pub transaction_time: u64,
#[serde(rename = "E")]
pub event_time: u64,
#[serde(rename = "st")]
pub symbol_type: SymbolType,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum Payload {
Tickers(Vec<TickerEvent>),
MarkPrices(Vec<MarkPriceEvent>),
AggTrade(AggTradeEvent),
Kline(KlineEvent),
MarkPrice(MarkPriceEvent),
Ticker(TickerEvent),
PartialDepth(PartialDepthEvent),
BookTicker(BookTickerEvent),
Unknown {
event_type: String,
raw: String,
},
}
impl Serialize for Payload {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
match self {
Self::Tickers(rows) => rows.serialize(serializer),
Self::MarkPrices(rows) => rows.serialize(serializer),
Self::AggTrade(event) => event.serialize(serializer),
Self::Kline(event) => event.serialize(serializer),
Self::MarkPrice(event) => event.serialize(serializer),
Self::Ticker(event) => event.serialize(serializer),
Self::PartialDepth(event) => event.serialize(serializer),
Self::BookTicker(event) => event.serialize(serializer),
Self::Unknown { raw, .. } => serde_json::from_str::<Value>(raw)
.map_err(serde::ser::Error::custom)?
.serialize(serializer),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct Update {
pub stream: StreamName,
pub payload: Payload,
}
impl Serialize for Update {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeMap;
let mut map = serializer.serialize_map(Some(2))?;
map.serialize_entry("stream", &self.stream.to_string())?;
map.serialize_entry("data", &self.payload)?;
map.end()
}
}
#[derive(Deserialize)]
struct Envelope {
stream: String,
data: Value,
}
impl Update {
pub fn from_json(text: &str) -> Result<Self, UsdmWsError> {
let envelope: Envelope =
serde_json::from_str(text).map_err(|err| frame_error("", text, err))?;
let stream: StreamName = envelope
.stream
.parse()
.map_err(|err| frame_error(&envelope.stream, text, err))?;
let payload = decode(&stream, envelope.data)
.map_err(|err| frame_error(&envelope.stream, text, err))?;
Ok(Self { stream, payload })
}
}
fn frame_error(stream: &str, raw: &str, reason: impl fmt::Display) -> UsdmWsError {
UsdmWsError::Frame {
stream: stream.to_owned(),
raw: raw.to_owned(),
reason: reason.to_string(),
}
}
fn decode(stream: &StreamName, data: Value) -> Result<Payload, serde_json::Error> {
let expected = match stream {
StreamName::AllTickers => return serde_json::from_value(data).map(Payload::Tickers),
StreamName::AllMarkPrices => return serde_json::from_value(data).map(Payload::MarkPrices),
StreamName::AggTrade(_) => "aggTrade",
StreamName::Kline(..) => "kline",
StreamName::MarkPrice(_) => "markPriceUpdate",
StreamName::Ticker(_) => "24hrTicker",
StreamName::PartialDepth(..) => "depthUpdate",
StreamName::BookTicker(_) => "bookTicker",
};
let event_type = data.get("e").and_then(Value::as_str).unwrap_or_default();
if event_type != expected {
return Ok(Payload::Unknown {
event_type: event_type.to_owned(),
raw: data.to_string(),
});
}
Ok(match stream {
StreamName::AggTrade(_) => Payload::AggTrade(serde_json::from_value(data)?),
StreamName::Kline(..) => Payload::Kline(serde_json::from_value(data)?),
StreamName::MarkPrice(_) => Payload::MarkPrice(serde_json::from_value(data)?),
StreamName::Ticker(_) => Payload::Ticker(serde_json::from_value(data)?),
StreamName::PartialDepth(..) => Payload::PartialDepth(serde_json::from_value(data)?),
StreamName::BookTicker(_) => Payload::BookTicker(serde_json::from_value(data)?),
StreamName::AllTickers | StreamName::AllMarkPrices => unreachable!("returned above"),
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::usdm::ws::fixtures;
#[test]
fn every_captured_frame_decodes_as_its_stream() {
for (name, frame) in fixtures::ALL {
let update = Update::from_json(frame).unwrap_or_else(|e| panic!("{name}: {e}"));
assert!(
!matches!(update.payload, Payload::Unknown { .. }),
"{name} decoded as Unknown"
);
}
}
#[test]
fn the_mark_price_array_keeps_every_field_prader_reads() {
let update = Update::from_json(fixtures::ALL_MARK_PRICES).unwrap();
assert_eq!(update.stream, StreamName::AllMarkPrices);
let Payload::MarkPrices(rows) = update.payload else {
panic!("not MarkPrices");
};
assert_eq!(rows.len(), 2);
assert!(rows[0].mark_price > Decimal::ZERO);
assert!(rows[0].next_funding_time > 0);
assert_eq!(rows[0].symbol_type, SymbolType::Um);
}
#[test]
fn each_field_holds_the_value_of_its_own_key() {
use serde_json::Value;
fn data(frame: &str) -> Value {
serde_json::from_str::<Value>(frame).unwrap()["data"].clone()
}
fn dec(row: &Value, key: &str) -> Decimal {
row[key]
.as_str()
.unwrap_or_else(|| panic!("{key} is not a string"))
.parse()
.unwrap()
}
fn int(row: &Value, key: &str) -> u64 {
row[key]
.as_u64()
.unwrap_or_else(|| panic!("{key} is not a u64"))
}
fn payload(frame: &str) -> Payload {
Update::from_json(frame).unwrap().payload
}
fn ticker(e: &TickerEvent, row: &Value) {
assert_eq!(
[
e.price_change,
e.price_change_percent,
e.weighted_avg_price,
e.last_price,
e.last_quantity,
e.open_price,
e.high_price,
e.low_price,
e.volume,
e.quote_volume,
],
["p", "P", "w", "c", "Q", "o", "h", "l", "v", "q"].map(|k| dec(row, k)),
"{}",
e.symbol
);
assert_eq!(
[e.event_time, e.open_time, e.close_time, e.trade_count],
["E", "O", "C", "n"].map(|k| int(row, k)),
"{}",
e.symbol
);
}
fn mark(e: &MarkPriceEvent, row: &Value) {
assert_eq!(
[
e.mark_price,
e.mark_price_moving_average,
e.estimated_settle_price,
e.index_price,
e.funding_rate,
],
["p", "ap", "P", "i", "r"].map(|k| dec(row, k)),
"{}",
e.symbol
);
assert_eq!(
[e.event_time, e.next_funding_time],
["E", "T"].map(|k| int(row, k)),
"{}",
e.symbol
);
}
let Payload::Tickers(rows) = payload(fixtures::ALL_TICKERS) else {
panic!("not Tickers");
};
for (e, row) in rows
.iter()
.zip(data(fixtures::ALL_TICKERS).as_array().unwrap())
{
ticker(e, row);
}
let Payload::Ticker(e) = payload(fixtures::TICKER) else {
panic!("not Ticker");
};
ticker(&e, &data(fixtures::TICKER));
let Payload::MarkPrices(rows) = payload(fixtures::ALL_MARK_PRICES) else {
panic!("not MarkPrices");
};
for (e, row) in rows
.iter()
.zip(data(fixtures::ALL_MARK_PRICES).as_array().unwrap())
{
mark(e, row);
}
let Payload::MarkPrice(e) = payload(fixtures::MARK_PRICE) else {
panic!("not MarkPrice");
};
mark(&e, &data(fixtures::MARK_PRICE));
let row = data(fixtures::AGG_TRADE);
let Payload::AggTrade(e) = payload(fixtures::AGG_TRADE) else {
panic!("not AggTrade");
};
assert_eq!(
[e.price, e.quantity, e.normal_quantity],
["p", "q", "nq"].map(|k| dec(&row, k))
);
assert_eq!(
[
e.event_time,
e.id,
e.first_trade_id,
e.last_trade_id,
e.trade_time
],
["E", "a", "f", "l", "T"].map(|k| int(&row, k))
);
let row = data(fixtures::KLINE)["k"].clone();
let Payload::Kline(e) = payload(fixtures::KLINE) else {
panic!("not Kline");
};
let k = &e.kline;
assert_eq!(
[
k.open,
k.close,
k.high,
k.low,
k.volume,
k.quote_volume,
k.taker_buy_base_volume,
k.taker_buy_quote_volume,
],
["o", "c", "h", "l", "v", "q", "V", "Q"].map(|key| dec(&row, key))
);
assert_eq!(
[k.open_time, k.close_time, k.trade_count],
["t", "T", "n"].map(|key| int(&row, key))
);
let row = data(fixtures::PARTIAL_DEPTH);
let Payload::PartialDepth(e) = payload(fixtures::PARTIAL_DEPTH) else {
panic!("not PartialDepth");
};
assert_eq!(
[
e.event_time,
e.transaction_time,
e.first_update_id,
e.final_update_id,
e.previous_final_update_id,
],
["E", "T", "U", "u", "pu"].map(|k| int(&row, k))
);
for (side, key) in [(&e.bids, "b"), (&e.asks, "a")] {
let sent: Vec<(Decimal, Decimal)> = row[key]
.as_array()
.unwrap()
.iter()
.map(|level| {
(
level[0].as_str().unwrap().parse().unwrap(),
level[1].as_str().unwrap().parse().unwrap(),
)
})
.collect();
let decoded: Vec<(Decimal, Decimal)> =
side.iter().map(|l| (l.price, l.quantity)).collect();
assert_eq!(decoded, sent, "{key}");
}
let row = data(fixtures::BOOK_TICKER);
let Payload::BookTicker(e) = payload(fixtures::BOOK_TICKER) else {
panic!("not BookTicker");
};
assert_eq!(
[e.bid_price, e.bid_quantity, e.ask_price, e.ask_quantity],
["b", "B", "a", "A"].map(|k| dec(&row, k))
);
assert_eq!(
[e.update_id, e.transaction_time, e.event_time],
["u", "T", "E"].map(|k| int(&row, k))
);
}
#[test]
fn a_coin_m_row_and_an_unknown_symbol_type_both_decode() {
assert_eq!(
serde_json::from_str::<SymbolType>("2").unwrap(),
SymbolType::Cm
);
assert_eq!(
serde_json::from_str::<SymbolType>("7").unwrap(),
SymbolType::Other(7)
);
assert_eq!(serde_json::to_string(&SymbolType::Other(7)).unwrap(), "7");
}
#[test]
fn a_renamed_event_is_kept_whole_and_a_broken_one_is_a_frame_error() {
let renamed = r#"{"stream":"btcusdt@aggTrade","data":{"e":"aggTradeV2","x":1}}"#;
let update = Update::from_json(renamed).unwrap();
assert!(matches!(
&update.payload,
Payload::Unknown { event_type, .. } if event_type == "aggTradeV2"
));
let broken = r#"{"stream":"btcusdt@aggTrade","data":{"e":"aggTrade","p":"x"}}"#;
let err = Update::from_json(broken).unwrap_err();
assert!(
matches!(&err, UsdmWsError::Frame { stream, .. } if stream == "btcusdt@aggTrade"),
"{err:?}"
);
let unbuilt = r#"{"stream":"btcusdt@forceOrder","data":{}}"#;
assert!(matches!(
Update::from_json(unbuilt),
Err(UsdmWsError::Frame { .. })
));
assert!(matches!(
Update::from_json("not json"),
Err(UsdmWsError::Frame { .. })
));
}
#[test]
fn an_update_serialises_back_to_its_envelope() {
let update = Update::from_json(fixtures::BOOK_TICKER).unwrap();
let again: Value = serde_json::to_value(&update).unwrap();
let wire: Value = serde_json::from_str(fixtures::BOOK_TICKER).unwrap();
assert_eq!(again, wire);
}
}