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}
91
92/// Declarative subscription request — built up fluently, consumed by
93/// [`crate::Station::subscribe`].
94#[derive(Debug, Default, Clone)]
95pub struct SubscriptionSet {
96    pub(crate) entries: Vec<Entry>,
97}
98
99impl SubscriptionSet {
100    pub fn new() -> Self { Self::default() }
101
102    pub fn add(
103        mut self,
104        exchange: ExchangeId,
105        symbol: impl Into<String>,
106        account_type: AccountType,
107        streams: impl IntoIterator<Item = Stream>,
108    ) -> Self {
109        self.entries.push(Entry {
110            exchange,
111            symbol: symbol.into(),
112            account_type,
113            streams: streams.into_iter().collect(),
114        });
115        self
116    }
117
118    pub fn len(&self) -> usize { self.entries.len() }
119    pub fn is_empty(&self) -> bool { self.entries.is_empty() }
120}
121
122/// Events forwarded to consumers. One variant per market-data class.
123#[derive(Debug, Clone)]
124pub enum Event {
125    Trade {
126        exchange: ExchangeId,
127        symbol: String,
128        point: TradePoint,
129    },
130    AggTrade {
131        exchange: ExchangeId,
132        symbol: String,
133        point: AggTradePoint,
134    },
135    Bar {
136        exchange: ExchangeId,
137        symbol: String,
138        timeframe: KlineInterval,
139        point: BarPoint,
140    },
141    Ticker {
142        exchange: ExchangeId,
143        symbol: String,
144        point: TickerPoint,
145    },
146    OrderbookSnapshot {
147        exchange: ExchangeId,
148        symbol: String,
149        point: ObSnapshotPoint,
150    },
151    MarkPrice {
152        exchange: ExchangeId,
153        symbol: String,
154        point: MarkPricePoint,
155    },
156    FundingRate {
157        exchange: ExchangeId,
158        symbol: String,
159        point: FundingRatePoint,
160    },
161    OpenInterest {
162        exchange: ExchangeId,
163        symbol: String,
164        point: OpenInterestPoint,
165    },
166    Liquidation {
167        exchange: ExchangeId,
168        symbol: String,
169        point: LiquidationPoint,
170    },
171    // --- extended stream types ---
172    BlockTrade {
173        exchange: ExchangeId,
174        symbol: String,
175        point: BlockTradePoint,
176    },
177    IndexPrice {
178        exchange: ExchangeId,
179        symbol: String,
180        point: IndexPricePoint,
181    },
182    CompositeIndex {
183        exchange: ExchangeId,
184        symbol: String,
185        point: CompositeIndexPoint,
186    },
187    OptionGreeks {
188        exchange: ExchangeId,
189        symbol: String,
190        point: OptionGreeksPoint,
191    },
192    VolatilityIndex {
193        exchange: ExchangeId,
194        symbol: String,
195        point: VolatilityIndexPoint,
196    },
197    HistoricalVolatility {
198        exchange: ExchangeId,
199        symbol: String,
200        point: HistoricalVolatilityPoint,
201    },
202    Basis {
203        exchange: ExchangeId,
204        symbol: String,
205        point: BasisPoint,
206    },
207    InsuranceFund {
208        exchange: ExchangeId,
209        symbol: String,
210        point: InsuranceFundPoint,
211    },
212    OrderbookL3 {
213        exchange: ExchangeId,
214        symbol: String,
215        point: OrderbookL3Point,
216    },
217    SettlementEvent {
218        exchange: ExchangeId,
219        symbol: String,
220        point: SettlementEventPoint,
221    },
222    AuctionEvent {
223        exchange: ExchangeId,
224        symbol: String,
225        point: AuctionEventPoint,
226    },
227    MarketWarning {
228        exchange: ExchangeId,
229        symbol: String,
230        point: MarketWarningPoint,
231    },
232    RiskLimit {
233        exchange: ExchangeId,
234        symbol: String,
235        point: RiskLimitPoint,
236    },
237    PredictedFunding {
238        exchange: ExchangeId,
239        symbol: String,
240        point: PredictedFundingPoint,
241    },
242    FundingSettlement {
243        exchange: ExchangeId,
244        symbol: String,
245        point: FundingSettlementPoint,
246    },
247    MarkPriceKline {
248        exchange: ExchangeId,
249        symbol: String,
250        timeframe: KlineInterval,
251        point: MarkPriceKlinePoint,
252    },
253    IndexPriceKline {
254        exchange: ExchangeId,
255        symbol: String,
256        timeframe: KlineInterval,
257        point: IndexPriceKlinePoint,
258    },
259    PremiumIndexKline {
260        exchange: ExchangeId,
261        symbol: String,
262        timeframe: KlineInterval,
263        point: PremiumIndexKlinePoint,
264    },
265}
266
267impl Event {
268    pub fn exchange(&self) -> ExchangeId {
269        match self {
270            Event::Trade { exchange, .. } | Event::AggTrade { exchange, .. } |
271            Event::Bar { exchange, .. } | Event::Ticker { exchange, .. } |
272            Event::OrderbookSnapshot { exchange, .. } | Event::MarkPrice { exchange, .. } |
273            Event::FundingRate { exchange, .. } | Event::OpenInterest { exchange, .. } |
274            Event::Liquidation { exchange, .. } |
275            Event::BlockTrade { exchange, .. } | Event::IndexPrice { exchange, .. } |
276            Event::CompositeIndex { exchange, .. } | Event::OptionGreeks { exchange, .. } |
277            Event::VolatilityIndex { exchange, .. } | Event::HistoricalVolatility { exchange, .. } |
278            Event::Basis { exchange, .. } | Event::InsuranceFund { exchange, .. } |
279            Event::OrderbookL3 { exchange, .. } | Event::SettlementEvent { exchange, .. } |
280            Event::AuctionEvent { exchange, .. } | Event::MarketWarning { exchange, .. } |
281            Event::RiskLimit { exchange, .. } | Event::PredictedFunding { exchange, .. } |
282            Event::FundingSettlement { exchange, .. } |
283            Event::MarkPriceKline { exchange, .. } | Event::IndexPriceKline { exchange, .. } |
284            Event::PremiumIndexKline { exchange, .. } => *exchange,
285        }
286    }
287    pub fn symbol(&self) -> &str {
288        match self {
289            Event::Trade { symbol, .. } | Event::AggTrade { symbol, .. } |
290            Event::Bar { symbol, .. } | Event::Ticker { symbol, .. } |
291            Event::OrderbookSnapshot { symbol, .. } | Event::MarkPrice { symbol, .. } |
292            Event::FundingRate { symbol, .. } | Event::OpenInterest { symbol, .. } |
293            Event::Liquidation { symbol, .. } |
294            Event::BlockTrade { symbol, .. } | Event::IndexPrice { symbol, .. } |
295            Event::CompositeIndex { symbol, .. } | Event::OptionGreeks { symbol, .. } |
296            Event::VolatilityIndex { symbol, .. } | Event::HistoricalVolatility { symbol, .. } |
297            Event::Basis { symbol, .. } | Event::InsuranceFund { symbol, .. } |
298            Event::OrderbookL3 { symbol, .. } | Event::SettlementEvent { symbol, .. } |
299            Event::AuctionEvent { symbol, .. } | Event::MarketWarning { symbol, .. } |
300            Event::RiskLimit { symbol, .. } | Event::PredictedFunding { symbol, .. } |
301            Event::FundingSettlement { symbol, .. } |
302            Event::MarkPriceKline { symbol, .. } | Event::IndexPriceKline { symbol, .. } |
303            Event::PremiumIndexKline { symbol, .. } => symbol,
304        }
305    }
306
307    /// Replace the symbol label on this event in-place.
308    ///
309    /// Used by `Station::subscribe` so each `SubscriptionHandle` sees the
310    /// user-input symbol it passed in `SubscriptionSet::add(...)`, regardless
311    /// of which other consumer first established the underlying multiplex.
312    /// The routing key (raw exchange-native) is unaffected; this only changes
313    /// the cosmetic label that `Event.symbol()` returns to the consumer.
314    pub(crate) fn set_symbol(&mut self, new_symbol: String) {
315        match self {
316            Event::Trade { symbol, .. }
317            | Event::AggTrade { symbol, .. }
318            | Event::Bar { symbol, .. }
319            | Event::Ticker { symbol, .. }
320            | Event::OrderbookSnapshot { symbol, .. }
321            | Event::MarkPrice { symbol, .. }
322            | Event::FundingRate { symbol, .. }
323            | Event::OpenInterest { symbol, .. }
324            | Event::Liquidation { symbol, .. }
325            | Event::BlockTrade { symbol, .. }
326            | Event::IndexPrice { symbol, .. }
327            | Event::CompositeIndex { symbol, .. }
328            | Event::OptionGreeks { symbol, .. }
329            | Event::VolatilityIndex { symbol, .. }
330            | Event::HistoricalVolatility { symbol, .. }
331            | Event::Basis { symbol, .. }
332            | Event::InsuranceFund { symbol, .. }
333            | Event::OrderbookL3 { symbol, .. }
334            | Event::SettlementEvent { symbol, .. }
335            | Event::AuctionEvent { symbol, .. }
336            | Event::MarketWarning { symbol, .. }
337            | Event::RiskLimit { symbol, .. }
338            | Event::PredictedFunding { symbol, .. }
339            | Event::FundingSettlement { symbol, .. }
340            | Event::MarkPriceKline { symbol, .. }
341            | Event::IndexPriceKline { symbol, .. }
342            | Event::PremiumIndexKline { symbol, .. } => *symbol = new_symbol,
343        }
344    }
345    pub fn timestamp_ms(&self) -> i64 {
346        use crate::series::DataPoint;
347        match self {
348            Event::Trade { point, .. } => point.timestamp_ms(),
349            Event::AggTrade { point, .. } => point.timestamp_ms(),
350            Event::Bar { point, .. } => point.timestamp_ms(),
351            Event::Ticker { point, .. } => point.timestamp_ms(),
352            Event::OrderbookSnapshot { point, .. } => point.timestamp_ms(),
353            Event::MarkPrice { point, .. } => point.timestamp_ms(),
354            Event::FundingRate { point, .. } => point.timestamp_ms(),
355            Event::OpenInterest { point, .. } => point.timestamp_ms(),
356            Event::Liquidation { point, .. } => point.timestamp_ms(),
357            Event::BlockTrade { point, .. } => point.timestamp_ms(),
358            Event::IndexPrice { point, .. } => point.timestamp_ms(),
359            Event::CompositeIndex { point, .. } => point.timestamp_ms(),
360            Event::OptionGreeks { point, .. } => point.timestamp_ms(),
361            Event::VolatilityIndex { point, .. } => point.timestamp_ms(),
362            Event::HistoricalVolatility { point, .. } => point.timestamp_ms(),
363            Event::Basis { point, .. } => point.timestamp_ms(),
364            Event::InsuranceFund { point, .. } => point.timestamp_ms(),
365            Event::OrderbookL3 { point, .. } => point.timestamp_ms(),
366            Event::SettlementEvent { point, .. } => point.timestamp_ms(),
367            Event::AuctionEvent { point, .. } => point.timestamp_ms(),
368            Event::MarketWarning { point, .. } => point.timestamp_ms(),
369            Event::RiskLimit { point, .. } => point.timestamp_ms(),
370            Event::PredictedFunding { point, .. } => point.timestamp_ms(),
371            Event::FundingSettlement { point, .. } => point.timestamp_ms(),
372            Event::MarkPriceKline { point, .. } => point.timestamp_ms(),
373            Event::IndexPriceKline { point, .. } => point.timestamp_ms(),
374            Event::PremiumIndexKline { point, .. } => point.timestamp_ms(),
375        }
376    }
377}
378
379/// RAII handle returned from `Station::subscribe`. Dropping releases the
380/// per-StreamKey consumer ref count; when count hits zero the multiplexer
381/// shuts down.
382pub struct SubscriptionHandle {
383    pub(crate) rx: tokio::sync::mpsc::UnboundedReceiver<Event>,
384    pub(crate) _refs: Vec<MultiplexRef>,
385}
386
387impl std::fmt::Debug for SubscriptionHandle {
388    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
389        f.debug_struct("SubscriptionHandle").finish()
390    }
391}
392
393impl SubscriptionHandle {
394    pub async fn recv(&mut self) -> Option<Event> {
395        self.rx.recv().await
396    }
397}
398
399pub(crate) struct MultiplexRef {
400    pub(crate) station: std::sync::Weak<crate::station::StationInner>,
401    pub(crate) key: SeriesKey,
402}
403
404impl Drop for MultiplexRef {
405    fn drop(&mut self) {
406        if let Some(inner) = self.station.upgrade() {
407            inner.release_consumer(&self.key);
408        }
409    }
410}
digdigdig3_station/subscription.rs

digdigdig3_station/
subscription.rs
