polyoxide_binance/usdm/ws/
stream.rs

1//! Stream names: the only way to name a USDⓈ-M market stream.
2//!
3//! Binance acknowledges every `SUBSCRIBE`, including a stream type that does
4//! not exist, an uppercase symbol, a spelling it does not serve (`@250ms`) and
5//! an unknown or delisted symbol, and none of those deliver anything. A name
6//! that will never deliver is indistinguishable from a quiet one, so names are
7//! built here, by construction, and never accepted as caller strings. The
8//! symbol is the one part this cannot check: `exchange().exchange_info()` lists
9//! the symbols that trade.
10
11use std::{fmt, str::FromStr};
12
13use crate::usdm::{
14    types::{Interval, Symbol},
15    ws::StreamPath,
16};
17
18/// Levels per side of a partial depth stream.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
20pub enum DepthLevels {
21    /// `depth5`.
22    Five,
23    /// `depth10`.
24    Ten,
25    /// `depth20`.
26    Twenty,
27}
28
29impl DepthLevels {
30    /// Every value, in order.
31    pub const ALL: &'static [Self] = &[Self::Five, Self::Ten, Self::Twenty];
32
33    fn as_str(self) -> &'static str {
34        match self {
35            Self::Five => "5",
36            Self::Ten => "10",
37            Self::Twenty => "20",
38        }
39    }
40}
41
42/// How often a partial depth stream pushes.
43#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
44pub enum DepthSpeed {
45    /// `@100ms`.
46    Ms100,
47    /// No suffix: Binance's default. An explicit `@250ms` is acknowledged and
48    /// delivers nothing (measured 2026-10-07).
49    Ms250,
50    /// `@500ms`.
51    Ms500,
52}
53
54impl DepthSpeed {
55    /// Every value, in order.
56    pub const ALL: &'static [Self] = &[Self::Ms100, Self::Ms250, Self::Ms500];
57
58    /// What follows the levels in a stream name.
59    fn suffix(self) -> &'static str {
60        match self {
61            Self::Ms100 => "@100ms",
62            Self::Ms250 => "",
63            Self::Ms500 => "@500ms",
64        }
65    }
66}
67
68/// A USDⓈ-M market stream.
69///
70/// `Display` renders the wire name, with the symbol's ASCII letters
71/// lowercased (`btcusdt@aggTrade`, `币安人生usdt@markPrice@1s`), and `FromStr`
72/// parses an echoed name back; [`Symbol::new`] uppercases ASCII, so the two
73/// round-trip.
74#[derive(Debug, Clone, PartialEq, Eq, Hash)]
75#[non_exhaustive]
76pub enum StreamName {
77    /// `!ticker@arr`: 24-hour tickers of every symbol that changed.
78    AllTickers,
79    /// `!markPrice@arr@1s`: mark price and funding of every symbol, each second.
80    AllMarkPrices,
81    /// `<s>@aggTrade`.
82    AggTrade(Symbol),
83    /// `<s>@kline_<interval>`.
84    Kline(Symbol, Interval),
85    /// `<s>@markPrice@1s`.
86    MarkPrice(Symbol),
87    /// `<s>@ticker`.
88    Ticker(Symbol),
89    /// `<s>@depth<levels>@<speed>`, on the `/public` path.
90    PartialDepth(Symbol, DepthLevels, DepthSpeed),
91    /// `<s>@bookTicker`, on the `/public` path.
92    BookTicker(Symbol),
93}
94
95impl StreamName {
96    /// The path whose connection carries this stream.
97    pub fn path(&self) -> StreamPath {
98        match self {
99            Self::PartialDepth(..) | Self::BookTicker(_) => StreamPath::Public,
100            Self::AllTickers
101            | Self::AllMarkPrices
102            | Self::AggTrade(_)
103            | Self::Kline(..)
104            | Self::MarkPrice(_)
105            | Self::Ticker(_) => StreamPath::Market,
106        }
107    }
108
109    /// The symbol, for a single-symbol stream.
110    pub fn symbol(&self) -> Option<&Symbol> {
111        match self {
112            Self::AllTickers | Self::AllMarkPrices => None,
113            Self::AggTrade(s)
114            | Self::Kline(s, _)
115            | Self::MarkPrice(s)
116            | Self::Ticker(s)
117            | Self::PartialDepth(s, ..)
118            | Self::BookTicker(s) => Some(s),
119        }
120    }
121}
122
123impl fmt::Display for StreamName {
124    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
125        let lower = |s: &Symbol| s.as_str().to_ascii_lowercase();
126        match self {
127            Self::AllTickers => f.write_str("!ticker@arr"),
128            Self::AllMarkPrices => f.write_str("!markPrice@arr@1s"),
129            Self::AggTrade(s) => write!(f, "{}@aggTrade", lower(s)),
130            Self::Kline(s, interval) => write!(f, "{}@kline_{interval}", lower(s)),
131            Self::MarkPrice(s) => write!(f, "{}@markPrice@1s", lower(s)),
132            Self::Ticker(s) => write!(f, "{}@ticker", lower(s)),
133            Self::PartialDepth(s, levels, speed) => {
134                write!(f, "{}@depth{}{}", lower(s), levels.as_str(), speed.suffix())
135            }
136            Self::BookTicker(s) => write!(f, "{}@bookTicker", lower(s)),
137        }
138    }
139}
140
141/// A string that is not a stream name this crate builds.
142#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
143#[error("{0:?} is not a USDⓈ-M stream name this crate supports: StreamName builds every one, with its symbol in lowercase")]
144pub struct InvalidStreamName(pub String);
145
146/// An `InvalidRequest`: the caller named a stream this crate does not build.
147impl polyoxide_venue::Classify for InvalidStreamName {
148    fn class(&self) -> polyoxide_venue::Class {
149        polyoxide_venue::Class::InvalidRequest
150    }
151}
152
153impl FromStr for StreamName {
154    type Err = InvalidStreamName;
155
156    fn from_str(name: &str) -> Result<Self, Self::Err> {
157        let invalid = || InvalidStreamName(name.to_owned());
158        match name {
159            "!ticker@arr" => return Ok(Self::AllTickers),
160            "!markPrice@arr@1s" => return Ok(Self::AllMarkPrices),
161            _ => {}
162        }
163        let (symbol, kind) = name.split_once('@').ok_or_else(invalid)?;
164        // A stream name spells its symbol in lowercase; one that does not was
165        // not built here and, on the wire, delivers nothing.
166        if symbol.chars().any(|c| c.is_ascii_uppercase()) {
167            return Err(invalid());
168        }
169        let symbol = Symbol::new(symbol).map_err(|_| invalid())?;
170        match kind {
171            "aggTrade" => Ok(Self::AggTrade(symbol)),
172            "markPrice@1s" => Ok(Self::MarkPrice(symbol)),
173            "ticker" => Ok(Self::Ticker(symbol)),
174            "bookTicker" => Ok(Self::BookTicker(symbol)),
175            _ => {
176                if let Some(interval) = kind.strip_prefix("kline_") {
177                    let interval = interval.parse().map_err(|_| invalid())?;
178                    return Ok(Self::Kline(symbol, interval));
179                }
180                let depth = kind.strip_prefix("depth").ok_or_else(invalid)?;
181                let (levels, speed) = DepthLevels::ALL
182                    .iter()
183                    .flat_map(|l| DepthSpeed::ALL.iter().map(move |s| (*l, *s)))
184                    .find(|(l, s)| depth.strip_prefix(l.as_str()) == Some(s.suffix()))
185                    .ok_or_else(invalid)?;
186                Ok(Self::PartialDepth(symbol, levels, speed))
187            }
188        }
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195
196    fn every_kind(symbol: &str) -> Vec<StreamName> {
197        let s = Symbol::new(symbol).unwrap();
198        let mut names = vec![
199            StreamName::AllTickers,
200            StreamName::AllMarkPrices,
201            StreamName::AggTrade(s.clone()),
202            StreamName::MarkPrice(s.clone()),
203            StreamName::Ticker(s.clone()),
204            StreamName::BookTicker(s.clone()),
205        ];
206        names.extend(
207            Interval::ALL
208                .iter()
209                .map(|i| StreamName::Kline(s.clone(), *i)),
210        );
211        for levels in DepthLevels::ALL {
212            for speed in DepthSpeed::ALL {
213                names.push(StreamName::PartialDepth(s.clone(), *levels, *speed));
214            }
215        }
216        names
217    }
218
219    #[test]
220    fn every_name_round_trips_through_its_wire_spelling() {
221        for symbol in [
222            "BTCUSDT",
223            "BTCUSDT_261225",
224            "币安人生USDT",
225            "1000PEPEUSDT",
226            "ÉTÉUSDT",
227        ] {
228            for name in every_kind(symbol) {
229                let wire = name.to_string();
230                assert_eq!(wire.parse::<StreamName>(), Ok(name.clone()), "{wire}");
231            }
232        }
233    }
234
235    #[test]
236    fn names_are_spelled_as_the_wire_spells_them() {
237        let btc = Symbol::new("BTCUSDT").unwrap();
238        let chinese = Symbol::new("币安人生USDT").unwrap();
239        let cases = [
240            (StreamName::AllTickers, "!ticker@arr"),
241            (StreamName::AllMarkPrices, "!markPrice@arr@1s"),
242            (StreamName::AggTrade(btc.clone()), "btcusdt@aggTrade"),
243            (
244                StreamName::Kline(btc.clone(), Interval::Mo1),
245                "btcusdt@kline_1M",
246            ),
247            (StreamName::MarkPrice(chinese), "币安人生usdt@markPrice@1s"),
248            (StreamName::Ticker(btc.clone()), "btcusdt@ticker"),
249            (
250                StreamName::PartialDepth(btc.clone(), DepthLevels::Twenty, DepthSpeed::Ms100),
251                "btcusdt@depth20@100ms",
252            ),
253            (StreamName::BookTicker(btc), "btcusdt@bookTicker"),
254        ];
255        for (name, wire) in cases {
256            assert_eq!(name.to_string(), wire);
257        }
258    }
259
260    #[test]
261    fn every_depth_name_is_one_binance_delivers() {
262        // Measured 2026-10-07: each of these delivered frames; the explicit
263        // `@250ms` spelling was acknowledged and delivered nothing.
264        let btc = Symbol::new("BTCUSDT").unwrap();
265        for (levels, speed, wire) in [
266            (DepthLevels::Five, DepthSpeed::Ms100, "btcusdt@depth5@100ms"),
267            (DepthLevels::Five, DepthSpeed::Ms250, "btcusdt@depth5"),
268            (DepthLevels::Five, DepthSpeed::Ms500, "btcusdt@depth5@500ms"),
269            (DepthLevels::Ten, DepthSpeed::Ms100, "btcusdt@depth10@100ms"),
270            (DepthLevels::Ten, DepthSpeed::Ms250, "btcusdt@depth10"),
271            (DepthLevels::Ten, DepthSpeed::Ms500, "btcusdt@depth10@500ms"),
272            (
273                DepthLevels::Twenty,
274                DepthSpeed::Ms100,
275                "btcusdt@depth20@100ms",
276            ),
277            (DepthLevels::Twenty, DepthSpeed::Ms250, "btcusdt@depth20"),
278            (
279                DepthLevels::Twenty,
280                DepthSpeed::Ms500,
281                "btcusdt@depth20@500ms",
282            ),
283        ] {
284            let name = StreamName::PartialDepth(btc.clone(), levels, speed);
285            assert_eq!(name.to_string(), wire);
286            assert_eq!(wire.parse::<StreamName>(), Ok(name));
287        }
288    }
289
290    #[test]
291    fn depth_and_book_ticker_ride_the_public_path_and_the_rest_the_market_path() {
292        for name in every_kind("BTCUSDT") {
293            let expected = match name {
294                StreamName::PartialDepth(..) | StreamName::BookTicker(_) => StreamPath::Public,
295                _ => StreamPath::Market,
296            };
297            assert_eq!(name.path(), expected, "{name}");
298        }
299    }
300
301    #[test]
302    fn a_name_this_crate_does_not_build_does_not_parse() {
303        for bad in [
304            "BTCUSDT@aggTrade",
305            "btcusdt@nonsense",
306            "btcusdt@depth7@100ms",
307            "btcusdt@depth5@250ms",
308            "btcusdt@depth50",
309            "btcusdt@kline_1s",
310            "btcusdt",
311            "!bookTicker",
312            "btc usdt@ticker",
313        ] {
314            assert_eq!(
315                bad.parse::<StreamName>(),
316                Err(InvalidStreamName(bad.to_owned())),
317                "{bad}"
318            );
319        }
320    }
321
322    #[test]
323    fn a_name_that_does_not_parse_is_an_invalid_request() {
324        use polyoxide_venue::{Class, Classify};
325
326        let err = "btcusdt".parse::<StreamName>().unwrap_err();
327        assert_eq!(err.class(), Class::InvalidRequest);
328        assert!(err.is_fault());
329        assert_eq!(err.retry_after(), None);
330        assert!(!err.is_retriable());
331    }
332}
polyoxide_binance/usdm/ws/stream.rs

polyoxide_binance/usdm/ws/
stream.rs
