digdigdig3_station/
subscription.rs

1use digdigdig3::core::types::{AccountType, ExchangeId};
2use digdigdig3::core::websocket::KlineInterval;
3
4use crate::data::{
5    AggTradePoint, AuctionEventPoint, BarPoint, BasisPoint, BlockTradePoint, CompositeIndexPoint,
6    FundingRatePoint, FundingSettlementPoint, HistoricalVolatilityPoint, IndexPriceKlinePoint,
7    IndexPricePoint, InsuranceFundPoint, LiquidationPoint, MarkPriceKlinePoint, MarkPricePoint,
8    MarketWarningPoint, ObDeltaPoint, ObSnapshotPoint, OpenInterestPoint, OptionGreeksPoint,
9    OrderbookL3Point, PredictedFundingPoint, PremiumIndexKlinePoint, RiskLimitPoint,
10    SettlementEventPoint, TickerPoint, TradePoint, VolatilityIndexPoint,
11};
12use crate::series::{Kind, SeriesKey};
13
14/// User-facing stream class to request in a `SubscriptionSet`.
15///
16/// `Kline` carries a typed `KlineInterval` (e.g. `KlineInterval::new("1m")`).
17/// All other variants are parameterless.
18#[derive(Debug, Clone, PartialEq, Eq, Hash)]
19pub enum Stream {
20    Ticker,
21    Trade,
22    Orderbook,
23    OrderbookDelta,
24    Kline(KlineInterval),
25    MarkPrice,
26    FundingRate,
27    OpenInterest,
28    Liquidation,
29    AggTrade,
30    // --- extended stream types ---
31    BlockTrade,
32    IndexPrice,
33    CompositeIndex,
34    OptionGreeks,
35    VolatilityIndex,
36    HistoricalVolatility,
37    Basis,
38    InsuranceFund,
39    OrderbookL3,
40    SettlementEvent,
41    AuctionEvent,
42    MarketWarning,
43    RiskLimit,
44    PredictedFunding,
45    FundingSettlement,
46    MarkPriceKline(KlineInterval),
47    IndexPriceKline(KlineInterval),
48    PremiumIndexKline(KlineInterval),
49}
50
51impl Stream {
52    pub(crate) fn to_kind(&self) -> Kind {
53        match self {
54            Stream::Trade => Kind::Trade,
55            Stream::AggTrade => Kind::AggTrade,
56            Stream::Kline(iv) => Kind::Kline(iv.clone()),
57            Stream::Ticker => Kind::Ticker,
58            Stream::Orderbook => Kind::Orderbook,
59            Stream::OrderbookDelta => Kind::OrderbookDelta,
60            Stream::MarkPrice => Kind::MarkPrice,
61            Stream::FundingRate => Kind::FundingRate,
62            Stream::OpenInterest => Kind::OpenInterest,
63            Stream::Liquidation => Kind::Liquidation,
64            Stream::BlockTrade => Kind::BlockTrade,
65            Stream::IndexPrice => Kind::IndexPrice,
66            Stream::CompositeIndex => Kind::CompositeIndex,
67            Stream::OptionGreeks => Kind::OptionGreeks,
68            Stream::VolatilityIndex => Kind::VolatilityIndex,
69            Stream::HistoricalVolatility => Kind::HistoricalVolatility,
70            Stream::Basis => Kind::Basis,
71            Stream::InsuranceFund => Kind::InsuranceFund,
72            Stream::OrderbookL3 => Kind::OrderbookL3,
73            Stream::SettlementEvent => Kind::SettlementEvent,
74            Stream::AuctionEvent => Kind::AuctionEvent,
75            Stream::MarketWarning => Kind::MarketWarning,
76            Stream::RiskLimit => Kind::RiskLimit,
77            Stream::PredictedFunding => Kind::PredictedFunding,
78            Stream::FundingSettlement => Kind::FundingSettlement,
79            Stream::MarkPriceKline(iv) => Kind::MarkPriceKline(iv.clone()),
80            Stream::IndexPriceKline(iv) => Kind::IndexPriceKline(iv.clone()),
81            Stream::PremiumIndexKline(iv) => Kind::PremiumIndexKline(iv.clone()),
82        }
83    }
84}
85
86#[derive(Debug, Clone)]
87pub(crate) struct Entry {
88    pub(crate) exchange: ExchangeId,
89    pub(crate) symbol: String,
90    pub(crate) account_type: AccountType,
91    pub(crate) streams: Vec<Stream>,
92    /// If true, `symbol` is the raw exchange-native string and must be
93    /// passed through to the WS connector verbatim — no `SymbolNormalizer`
94    /// translation. Set by `SubscriptionSet::add_raw`. Used for exotic
95    /// instrument IDs that don't fit the canonical BASE-QUOTE shape
96    /// (Deribit options like `BTC-23MAY26-86000-C`, exchange-specific
97    /// suffixes, etc.).
98    pub(crate) is_raw: bool,
99}
100
101/// Declarative subscription request — built up fluently, consumed by
102/// [`crate::Station::subscribe`].
103#[derive(Debug, Default, Clone)]
104pub struct SubscriptionSet {
105    pub(crate) entries: Vec<Entry>,
106}
107
108impl SubscriptionSet {
109    pub fn new() -> Self { Self::default() }
110
111    /// Add a subscription. `symbol` is canonical (e.g. `"BTC-USDT"`,
112    /// `"BTCUSDT"`, `"BTC/USDT"`) — it is parsed into a canonical
113    /// `Symbol` and translated to the exchange-native form via
114    /// `SymbolNormalizer`. Use [`Self::add_raw`] for instrument IDs that
115    /// don't fit the canonical BASE-QUOTE shape (Deribit options,
116    /// exchange-specific futures suffixes, etc.).
117    pub fn add(
118        mut self,
119        exchange: ExchangeId,
120        symbol: impl Into<String>,
121        account_type: AccountType,
122        streams: impl IntoIterator<Item = Stream>,
123    ) -> Self {
124        self.entries.push(Entry {
125            exchange,
126            symbol: symbol.into(),
127            account_type,
128            streams: streams.into_iter().collect(),
129            is_raw: false,
130        });
131        self
132    }
133
134    /// Add a subscription with a raw exchange-native symbol. `symbol` is
135    /// passed through to the connector verbatim — no `SymbolNormalizer`
136    /// translation. Use for instrument IDs that don't fit the canonical
137    /// BASE-QUOTE shape:
138    /// - Deribit options: `"BTC-23MAY26-86000-C"`
139    /// - Futures with date suffix: `"BTCUSDT_240329"`
140    /// - Index symbols: `".DEFI"`, `"BTCUSD-PERP"`
141    ///
142    /// The caller is responsible for using the exact wire format the
143    /// exchange expects — `Event.symbol` on the handle will mirror the
144    /// raw string back.
145    pub fn add_raw(
146        mut self,
147        exchange: ExchangeId,
148        symbol: impl Into<String>,
149        account_type: AccountType,
150        streams: impl IntoIterator<Item = Stream>,
151    ) -> Self {
152        self.entries.push(Entry {
153            exchange,
154            symbol: symbol.into(),
155            account_type,
156            streams: streams.into_iter().collect(),
157            is_raw: true,
158        });
159        self
160    }
161
162    pub fn len(&self) -> usize { self.entries.len() }
163    pub fn is_empty(&self) -> bool { self.entries.is_empty() }
164}
165
166/// Events forwarded to consumers. One variant per market-data class.
167#[derive(Debug, Clone)]
168pub enum Event {
169    Trade {
170        exchange: ExchangeId,
171        symbol: String,
172        point: TradePoint,
173    },
174    AggTrade {
175        exchange: ExchangeId,
176        symbol: String,
177        point: AggTradePoint,
178    },
179    Bar {
180        exchange: ExchangeId,
181        symbol: String,
182        timeframe: KlineInterval,
183        point: BarPoint,
184    },
185    Ticker {
186        exchange: ExchangeId,
187        symbol: String,
188        point: TickerPoint,
189    },
190    OrderbookSnapshot {
191        exchange: ExchangeId,
192        symbol: String,
193        point: ObSnapshotPoint,
194    },
195    OrderbookDelta {
196        exchange: ExchangeId,
197        symbol: String,
198        point: ObDeltaPoint,
199    },
200    MarkPrice {
201        exchange: ExchangeId,
202        symbol: String,
203        point: MarkPricePoint,
204    },
205    FundingRate {
206        exchange: ExchangeId,
207        symbol: String,
208        point: FundingRatePoint,
209    },
210    OpenInterest {
211        exchange: ExchangeId,
212        symbol: String,
213        point: OpenInterestPoint,
214    },
215    Liquidation {
216        exchange: ExchangeId,
217        symbol: String,
218        point: LiquidationPoint,
219    },
220    // --- extended stream types ---
221    BlockTrade {
222        exchange: ExchangeId,
223        symbol: String,
224        point: BlockTradePoint,
225    },
226    IndexPrice {
227        exchange: ExchangeId,
228        symbol: String,
229        point: IndexPricePoint,
230    },
231    CompositeIndex {
232        exchange: ExchangeId,
233        symbol: String,
234        point: CompositeIndexPoint,
235    },
236    OptionGreeks {
237        exchange: ExchangeId,
238        symbol: String,
239        point: OptionGreeksPoint,
240    },
241    VolatilityIndex {
242        exchange: ExchangeId,
243        symbol: String,
244        point: VolatilityIndexPoint,
245    },
246    HistoricalVolatility {
247        exchange: ExchangeId,
248        symbol: String,
249        point: HistoricalVolatilityPoint,
250    },
251    Basis {
252        exchange: ExchangeId,
253        symbol: String,
254        point: BasisPoint,
255    },
256    InsuranceFund {
257        exchange: ExchangeId,
258        symbol: String,
259        point: InsuranceFundPoint,
260    },
261    OrderbookL3 {
262        exchange: ExchangeId,
263        symbol: String,
264        point: OrderbookL3Point,
265    },
266    SettlementEvent {
267        exchange: ExchangeId,
268        symbol: String,
269        point: SettlementEventPoint,
270    },
271    AuctionEvent {
272        exchange: ExchangeId,
273        symbol: String,
274        point: AuctionEventPoint,
275    },
276    MarketWarning {
277        exchange: ExchangeId,
278        symbol: String,
279        point: MarketWarningPoint,
280    },
281    RiskLimit {
282        exchange: ExchangeId,
283        symbol: String,
284        point: RiskLimitPoint,
285    },
286    PredictedFunding {
287        exchange: ExchangeId,
288        symbol: String,
289        point: PredictedFundingPoint,
290    },
291    FundingSettlement {
292        exchange: ExchangeId,
293        symbol: String,
294        point: FundingSettlementPoint,
295    },
296    MarkPriceKline {
297        exchange: ExchangeId,
298        symbol: String,
299        timeframe: KlineInterval,
300        point: MarkPriceKlinePoint,
301    },
302    IndexPriceKline {
303        exchange: ExchangeId,
304        symbol: String,
305        timeframe: KlineInterval,
306        point: IndexPriceKlinePoint,
307    },
308    PremiumIndexKline {
309        exchange: ExchangeId,
310        symbol: String,
311        timeframe: KlineInterval,
312        point: PremiumIndexKlinePoint,
313    },
314}
315
316impl Event {
317    pub fn exchange(&self) -> ExchangeId {
318        match self {
319            Event::Trade { exchange, .. } | Event::AggTrade { exchange, .. } |
320            Event::Bar { exchange, .. } | Event::Ticker { exchange, .. } |
321            Event::OrderbookSnapshot { exchange, .. } | Event::OrderbookDelta { exchange, .. } |
322            Event::MarkPrice { exchange, .. } |
323            Event::FundingRate { exchange, .. } | Event::OpenInterest { exchange, .. } |
324            Event::Liquidation { exchange, .. } |
325            Event::BlockTrade { exchange, .. } | Event::IndexPrice { exchange, .. } |
326            Event::CompositeIndex { exchange, .. } | Event::OptionGreeks { exchange, .. } |
327            Event::VolatilityIndex { exchange, .. } | Event::HistoricalVolatility { exchange, .. } |
328            Event::Basis { exchange, .. } | Event::InsuranceFund { exchange, .. } |
329            Event::OrderbookL3 { exchange, .. } | Event::SettlementEvent { exchange, .. } |
330            Event::AuctionEvent { exchange, .. } | Event::MarketWarning { exchange, .. } |
331            Event::RiskLimit { exchange, .. } | Event::PredictedFunding { exchange, .. } |
332            Event::FundingSettlement { exchange, .. } |
333            Event::MarkPriceKline { exchange, .. } | Event::IndexPriceKline { exchange, .. } |
334            Event::PremiumIndexKline { exchange, .. } => *exchange,
335        }
336    }
337    pub fn symbol(&self) -> &str {
338        match self {
339            Event::Trade { symbol, .. } | Event::AggTrade { symbol, .. } |
340            Event::Bar { symbol, .. } | Event::Ticker { symbol, .. } |
341            Event::OrderbookSnapshot { symbol, .. } | Event::OrderbookDelta { symbol, .. } |
342            Event::MarkPrice { symbol, .. } |
343            Event::FundingRate { symbol, .. } | Event::OpenInterest { symbol, .. } |
344            Event::Liquidation { symbol, .. } |
345            Event::BlockTrade { symbol, .. } | Event::IndexPrice { symbol, .. } |
346            Event::CompositeIndex { symbol, .. } | Event::OptionGreeks { symbol, .. } |
347            Event::VolatilityIndex { symbol, .. } | Event::HistoricalVolatility { symbol, .. } |
348            Event::Basis { symbol, .. } | Event::InsuranceFund { symbol, .. } |
349            Event::OrderbookL3 { symbol, .. } | Event::SettlementEvent { symbol, .. } |
350            Event::AuctionEvent { symbol, .. } | Event::MarketWarning { symbol, .. } |
351            Event::RiskLimit { symbol, .. } | Event::PredictedFunding { symbol, .. } |
352            Event::FundingSettlement { symbol, .. } |
353            Event::MarkPriceKline { symbol, .. } | Event::IndexPriceKline { symbol, .. } |
354            Event::PremiumIndexKline { symbol, .. } => symbol,
355        }
356    }
357
358    /// Replace the symbol label on this event in-place.
359    ///
360    /// Used by `Station::subscribe` so each `SubscriptionHandle` sees the
361    /// user-input symbol it passed in `SubscriptionSet::add(...)`, regardless
362    /// of which other consumer first established the underlying multiplex.
363    /// The routing key (raw exchange-native) is unaffected; this only changes
364    /// the cosmetic label that `Event.symbol()` returns to the consumer.
365    pub(crate) fn set_symbol(&mut self, new_symbol: String) {
366        match self {
367            Event::Trade { symbol, .. }
368            | Event::AggTrade { symbol, .. }
369            | Event::Bar { symbol, .. }
370            | Event::Ticker { symbol, .. }
371            | Event::OrderbookSnapshot { symbol, .. }
372            | Event::OrderbookDelta { symbol, .. }
373            | Event::MarkPrice { symbol, .. }
374            | Event::FundingRate { symbol, .. }
375            | Event::OpenInterest { symbol, .. }
376            | Event::Liquidation { symbol, .. }
377            | Event::BlockTrade { symbol, .. }
378            | Event::IndexPrice { symbol, .. }
379            | Event::CompositeIndex { symbol, .. }
380            | Event::OptionGreeks { symbol, .. }
381            | Event::VolatilityIndex { symbol, .. }
382            | Event::HistoricalVolatility { symbol, .. }
383            | Event::Basis { symbol, .. }
384            | Event::InsuranceFund { symbol, .. }
385            | Event::OrderbookL3 { symbol, .. }
386            | Event::SettlementEvent { symbol, .. }
387            | Event::AuctionEvent { symbol, .. }
388            | Event::MarketWarning { symbol, .. }
389            | Event::RiskLimit { symbol, .. }
390            | Event::PredictedFunding { symbol, .. }
391            | Event::FundingSettlement { symbol, .. }
392            | Event::MarkPriceKline { symbol, .. }
393            | Event::IndexPriceKline { symbol, .. }
394            | Event::PremiumIndexKline { symbol, .. } => *symbol = new_symbol,
395        }
396    }
397    pub fn timestamp_ms(&self) -> i64 {
398        use crate::series::DataPoint;
399        match self {
400            Event::Trade { point, .. } => point.timestamp_ms(),
401            Event::AggTrade { point, .. } => point.timestamp_ms(),
402            Event::Bar { point, .. } => point.timestamp_ms(),
403            Event::Ticker { point, .. } => point.timestamp_ms(),
404            Event::OrderbookSnapshot { point, .. } => point.timestamp_ms(),
405            Event::OrderbookDelta { point, .. } => point.timestamp_ms(),
406            Event::MarkPrice { point, .. } => point.timestamp_ms(),
407            Event::FundingRate { point, .. } => point.timestamp_ms(),
408            Event::OpenInterest { point, .. } => point.timestamp_ms(),
409            Event::Liquidation { point, .. } => point.timestamp_ms(),
410            Event::BlockTrade { point, .. } => point.timestamp_ms(),
411            Event::IndexPrice { point, .. } => point.timestamp_ms(),
412            Event::CompositeIndex { point, .. } => point.timestamp_ms(),
413            Event::OptionGreeks { point, .. } => point.timestamp_ms(),
414            Event::VolatilityIndex { point, .. } => point.timestamp_ms(),
415            Event::HistoricalVolatility { point, .. } => point.timestamp_ms(),
416            Event::Basis { point, .. } => point.timestamp_ms(),
417            Event::InsuranceFund { point, .. } => point.timestamp_ms(),
418            Event::OrderbookL3 { point, .. } => point.timestamp_ms(),
419            Event::SettlementEvent { point, .. } => point.timestamp_ms(),
420            Event::AuctionEvent { point, .. } => point.timestamp_ms(),
421            Event::MarketWarning { point, .. } => point.timestamp_ms(),
422            Event::RiskLimit { point, .. } => point.timestamp_ms(),
423            Event::PredictedFunding { point, .. } => point.timestamp_ms(),
424            Event::FundingSettlement { point, .. } => point.timestamp_ms(),
425            Event::MarkPriceKline { point, .. } => point.timestamp_ms(),
426            Event::IndexPriceKline { point, .. } => point.timestamp_ms(),
427            Event::PremiumIndexKline { point, .. } => point.timestamp_ms(),
428        }
429    }
430}
431
432/// RAII handle returned from `Station::subscribe`. Dropping releases the
433/// per-StreamKey consumer ref count; when count hits zero the multiplexer
434/// shuts down.
435pub struct SubscriptionHandle {
436    pub(crate) rx: tokio::sync::mpsc::UnboundedReceiver<Event>,
437    pub(crate) _refs: Vec<MultiplexRef>,
438}
439
440impl std::fmt::Debug for SubscriptionHandle {
441    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
442        f.debug_struct("SubscriptionHandle").finish()
443    }
444}
445
446impl SubscriptionHandle {
447    pub async fn recv(&mut self) -> Option<Event> {
448        self.rx.recv().await
449    }
450}
451
452/// Per-stream subscribe failure reported by [`crate::Station::subscribe`].
453///
454/// Carries everything a consumer needs to log + skip without forcing them
455/// to parse a `Display` string. `error.is_not_supported()` distinguishes
456/// venue-doesn't-expose-this-stream (architectural, quiet) from transient
457/// failures (worth surfacing).
458#[derive(Debug)]
459pub struct FailedStream {
460    pub exchange: ExchangeId,
461    pub account_type: AccountType,
462    /// User-input symbol form (NOT the normalized exchange-native form).
463    pub symbol: String,
464    pub stream: Stream,
465    pub error: crate::StationError,
466}
467
468/// Outcome of [`crate::Station::subscribe`] in continue-on-error mode.
469///
470/// `handle` always exists and carries events for every stream in `ok`.
471/// `failed` is a per-stream list of subscribes that did not produce a
472/// live forwarder. The most common entry there is
473/// `StationError::StreamNotSupported` — the venue genuinely does not
474/// expose the requested stream on the WS wire. Other errors (transport,
475/// REST, symbol normalize) also land here so the consumer can log them
476/// without aborting the whole subscribe batch.
477///
478/// `failed` is empty on success — callers that want fail-fast semantics
479/// can simply `if !report.failed.is_empty() { return Err(...) }`.
480pub struct SubscribeReport {
481    pub handle: SubscriptionHandle,
482    pub ok: Vec<SeriesKey>,
483    pub failed: Vec<FailedStream>,
484}
485
486impl SubscribeReport {
487    /// True if every requested stream produced a live forwarder.
488    pub fn is_fully_ok(&self) -> bool { self.failed.is_empty() }
489    /// Convenience: total streams requested (`ok.len() + failed.len()`).
490    pub fn total(&self) -> usize { self.ok.len() + self.failed.len() }
491}
492
493impl std::fmt::Debug for SubscribeReport {
494    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
495        f.debug_struct("SubscribeReport")
496            .field("ok", &self.ok.len())
497            .field("failed", &self.failed.len())
498            .finish()
499    }
500}
501
502pub(crate) struct MultiplexRef {
503    pub(crate) station: std::sync::Weak<crate::station::StationInner>,
504    pub(crate) key: SeriesKey,
505}
506
507impl Drop for MultiplexRef {
508    fn drop(&mut self) {
509        if let Some(inner) = self.station.upgrade() {
510            inner.release_consumer(&self.key);
511        }
512    }
513}
digdigdig3_station/subscription.rs

digdigdig3_station/
subscription.rs
