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

digdigdig3_station/
subscription.rs
