flowsurface_data/
stream.rs

1use exchange::{PushFrequency, Ticker, TickerInfo, Timeframe, adapter::StreamKind};
2use serde::{Deserialize, Serialize};
3
4#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
5pub enum PersistStreamKind {
6    Kline {
7        ticker: Ticker,
8        timeframe: Timeframe,
9    },
10    Depth(PersistDepth),
11    Trades {
12        ticker: Ticker,
13    },
14    /// Deprecated combined stream, kept for backward compatibility.
15    /// Will be converted to separate Depth and Trades on load.
16    DepthAndTrades(PersistDepth),
17}
18
19impl PersistStreamKind {
20    pub fn exchange(&self) -> exchange::adapter::Exchange {
21        match self {
22            PersistStreamKind::Kline { ticker, .. } => ticker.exchange,
23            PersistStreamKind::Depth(d) => d.ticker.exchange,
24            PersistStreamKind::Trades { ticker } => ticker.exchange,
25            PersistStreamKind::DepthAndTrades(d) => d.ticker.exchange,
26        }
27    }
28}
29
30#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
31pub struct PersistDepth {
32    pub ticker: Ticker,
33    #[serde(default = "default_depth_aggr")]
34    pub depth_aggr: exchange::adapter::StreamTicksize,
35    #[serde(default = "default_push_freq")]
36    pub push_freq: PushFrequency,
37}
38
39impl From<StreamKind> for PersistStreamKind {
40    fn from(stream: StreamKind) -> Self {
41        match stream {
42            StreamKind::Kline {
43                ticker_info,
44                timeframe,
45            } => PersistStreamKind::Kline {
46                ticker: ticker_info.ticker,
47                timeframe,
48            },
49            StreamKind::Depth {
50                ticker_info,
51                depth_aggr,
52                push_freq,
53            } => PersistStreamKind::Depth(PersistDepth {
54                ticker: ticker_info.ticker,
55                depth_aggr,
56                push_freq,
57            }),
58            StreamKind::Trades { ticker_info } => PersistStreamKind::Trades {
59                ticker: ticker_info.ticker,
60            },
61        }
62    }
63}
64
65impl PersistStreamKind {
66    /// Try to convert into runtime StreamKind list. `resolver` should return Some(TickerInfo) for a ticker,
67    /// otherwise the conversion fails (so caller can trigger a refresh / fetch).
68    pub fn into_stream_kinds<F>(self, mut resolver: F) -> Result<Vec<StreamKind>, String>
69    where
70        F: FnMut(&Ticker) -> Option<TickerInfo>,
71    {
72        match self {
73            PersistStreamKind::Kline { ticker, timeframe } => resolver(&ticker)
74                .map(|ti| {
75                    vec![StreamKind::Kline {
76                        ticker_info: ti,
77                        timeframe,
78                    }]
79                })
80                .ok_or_else(|| format!("Ticker metadata not found for {}", ticker)),
81            PersistStreamKind::Depth(d) => resolver(&d.ticker)
82                .map(|ti| {
83                    vec![StreamKind::Depth {
84                        ticker_info: ti,
85                        depth_aggr: d.depth_aggr,
86                        push_freq: d.push_freq,
87                    }]
88                })
89                .ok_or_else(|| format!("Ticker metadata not found for {}", d.ticker)),
90            PersistStreamKind::Trades { ticker } => resolver(&ticker)
91                .map(|ti| vec![StreamKind::Trades { ticker_info: ti }])
92                .ok_or_else(|| format!("Ticker metadata not found for {}", ticker)),
93            PersistStreamKind::DepthAndTrades(d) => resolver(&d.ticker)
94                .map(|ti| {
95                    vec![
96                        StreamKind::Depth {
97                            ticker_info: ti,
98                            depth_aggr: d.depth_aggr,
99                            push_freq: d.push_freq,
100                        },
101                        StreamKind::Trades { ticker_info: ti },
102                    ]
103                })
104                .ok_or_else(|| format!("Ticker metadata not found for {}", d.ticker)),
105        }
106    }
107}
108
109fn default_depth_aggr() -> exchange::adapter::StreamTicksize {
110    exchange::adapter::StreamTicksize::Client
111}
112
113fn default_push_freq() -> PushFrequency {
114    PushFrequency::ServerDefault
115}
flowsurface_data/stream.rs

flowsurface_data/
stream.rs
