digdigdig3_station/
subscription.rs

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

digdigdig3_station/
subscription.rs
