flowsurface_data/panel/
timesales.rs

1use std::{collections::VecDeque, time::Duration};
2
3use exchange::{
4    SizeUnit, TickerInfo, Trade, UnixMs,
5    unit::{Qty, qty::volume_size_unit},
6};
7use serde::{Deserialize, Serialize};
8
9use crate::util::ok_or_default;
10
11const TRADE_RETENTION_MS: u64 = 120_000;
12
13#[derive(Debug, Copy, Clone, PartialEq, Serialize, Deserialize)]
14pub struct Config {
15    #[serde(deserialize_with = "ok_or_default", default)]
16    pub ticker_display: TickerDisplay,
17    #[serde(deserialize_with = "ok_or_default", default)]
18    pub time_format: TimeFormat,
19    pub trade_size_filter: f32,
20    #[serde(default = "default_buffer_filter")]
21    pub trade_retention: Duration,
22    #[serde(deserialize_with = "ok_or_default", default)]
23    pub stacked_bar: Option<StackedBar>,
24}
25
26impl Default for Config {
27    fn default() -> Self {
28        Config {
29            ticker_display: TickerDisplay::default(),
30            time_format: TimeFormat::default(),
31            trade_size_filter: 0.0,
32            trade_retention: Duration::from_millis(TRADE_RETENTION_MS),
33            stacked_bar: StackedBar::Compact(StackedBarRatio::default()).into(),
34        }
35    }
36}
37
38#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
39pub enum TickerDisplay {
40    #[default]
41    Icon,
42    IconAndTicker,
43}
44
45#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
46pub enum TimeFormat {
47    #[default]
48    MinutesSecondsMillis,
49    HoursMinutesSecondsMillis,
50    HoursMinutesSeconds,
51}
52
53impl TimeFormat {
54    pub const ALL: [Self; 3] = [
55        Self::MinutesSecondsMillis,
56        Self::HoursMinutesSecondsMillis,
57        Self::HoursMinutesSeconds,
58    ];
59
60    pub fn pattern(self) -> &'static str {
61        match self {
62            Self::MinutesSecondsMillis => "%M:%S.%3f",
63            Self::HoursMinutesSecondsMillis => "%H:%M:%S.%3f",
64            Self::HoursMinutesSeconds => "%H:%M:%S",
65        }
66    }
67}
68
69impl std::fmt::Display for TimeFormat {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        f.write_str(match self {
72            Self::MinutesSecondsMillis => "mm:ss.ms",
73            Self::HoursMinutesSecondsMillis => "hh:mm:ss.ms",
74            Self::HoursMinutesSeconds => "hh:mm:ss",
75        })
76    }
77}
78
79fn default_buffer_filter() -> Duration {
80    Duration::from_millis(TRADE_RETENTION_MS)
81}
82
83#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Copy)]
84pub enum StackedBar {
85    Compact(StackedBarRatio),
86    Full(StackedBarRatio),
87}
88
89impl StackedBar {
90    pub fn ratio(self) -> StackedBarRatio {
91        match self {
92            StackedBar::Compact(r) | StackedBar::Full(r) => r,
93        }
94    }
95
96    pub fn with_ratio(self, r: StackedBarRatio) -> Self {
97        match self {
98            StackedBar::Compact(_) => StackedBar::Compact(r),
99            StackedBar::Full(_) => StackedBar::Full(r),
100        }
101    }
102}
103
104#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Default, Copy)]
105pub enum StackedBarRatio {
106    Count,
107    #[default]
108    Volume,
109    AverageSize,
110}
111
112impl std::fmt::Display for StackedBarRatio {
113    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114        match self {
115            StackedBarRatio::Count => write!(f, "Count"),
116            StackedBarRatio::AverageSize => write!(f, "Average trade size"),
117            StackedBarRatio::Volume => write!(f, "Volume"),
118        }
119    }
120}
121
122impl StackedBarRatio {
123    pub const ALL: [StackedBarRatio; 3] = [
124        StackedBarRatio::Count,
125        StackedBarRatio::Volume,
126        StackedBarRatio::AverageSize,
127    ];
128}
129
130#[derive(Clone, Copy)]
131pub struct TapeTrade {
132    trade: Trade,
133    ticker_info: TickerInfo,
134    notional: Qty,
135}
136
137impl std::ops::Deref for TapeTrade {
138    type Target = Trade;
139
140    fn deref(&self) -> &Trade {
141        &self.trade
142    }
143}
144
145impl TapeTrade {
146    pub fn new(ticker_info: TickerInfo, trade: Trade, size_unit: SizeUnit) -> Self {
147        let notional = Qty::from_f64(ticker_info.market_type().qty_in_quote_value(
148            trade.qty,
149            trade.price,
150            size_unit == SizeUnit::Quote,
151        ));
152        Self {
153            trade,
154            ticker_info,
155            notional,
156        }
157    }
158
159    pub fn ticker_info(&self) -> TickerInfo {
160        self.ticker_info
161    }
162
163    pub fn quantity_text(&self, multiple_tickers: bool) -> String {
164        format!(
165            "{}{}",
166            self.quantity_prefix(multiple_tickers, volume_size_unit()),
167            crate::util::abbr_large_numbers(self.quantity(multiple_tickers).to_f64())
168        )
169    }
170
171    pub fn quantity(&self, multiple_tickers: bool) -> Qty {
172        if multiple_tickers {
173            self.notional
174        } else {
175            self.qty
176        }
177    }
178
179    pub fn passes_filter(&self, threshold: f32) -> bool {
180        self.notional.to_f32_lossy() >= threshold
181    }
182
183    pub fn quantity_prefix(&self, multiple_tickers: bool, size_unit: SizeUnit) -> &'static str {
184        if multiple_tickers
185            || size_unit == SizeUnit::Quote
186            || self.ticker_info.market_type() == exchange::adapter::MarketKind::InversePerps
187        {
188            "$"
189        } else {
190            ""
191        }
192    }
193
194    pub fn merge_into(target: &mut VecDeque<Self>, mut incoming: Vec<Self>) {
195        if incoming.is_empty() {
196            return;
197        }
198        incoming.sort_by_key(|entry| entry.time);
199        if target
200            .back()
201            .is_none_or(|last| last.time <= incoming[0].time)
202        {
203            target.extend(incoming);
204            return;
205        }
206        let merge_start = target.partition_point(|entry| entry.time <= incoming[0].time);
207        let mut previous = target.split_off(merge_start);
208        for entry in incoming {
209            while previous.front().is_some_and(|old| old.time <= entry.time) {
210                target.push_back(previous.pop_front().unwrap());
211            }
212            target.push_back(entry);
213        }
214        target.append(&mut previous);
215    }
216}
217
218pub fn prune_trades(trades: &mut VecDeque<TapeTrade>, retention: Duration, now: UnixMs) -> bool {
219    let retention_ms = retention.as_millis() as u64;
220    let prune_slack_ms = retention_ms / 10;
221    let retention_cutoff = now.saturating_sub(retention_ms);
222    let slack_cutoff = now.saturating_sub(retention_ms.saturating_add(prune_slack_ms));
223
224    if trades
225        .front()
226        .is_none_or(|oldest| oldest.time >= slack_cutoff)
227    {
228        return false;
229    }
230
231    let previous_len = trades.len();
232    while trades
233        .front()
234        .is_some_and(|trade| trade.time < retention_cutoff)
235    {
236        trades.pop_front();
237    }
238    trades.len() != previous_len
239}
240
241#[derive(Debug, Clone, Copy)]
242pub enum HistAggValues {
243    Count { buy: u64, sell: u64 },
244    Qty { buy: Qty, sell: Qty },
245}
246
247#[derive(Default)]
248pub struct HistAgg {
249    buy_count: u64,
250    sell_count: u64,
251    buy_sum: Qty,
252    sell_sum: Qty,
253}
254
255impl HistAgg {
256    fn average_qty(sum: Qty, count: u64) -> Qty {
257        if count == 0 {
258            return Qty::ZERO;
259        }
260
261        let c = count.min(i64::MAX as u64) as i64;
262        let half = c / 2;
263        let rounded = if sum.units >= 0 {
264            sum.units.saturating_add(half).div_euclid(c)
265        } else {
266            sum.units.saturating_sub(half).div_euclid(c)
267        };
268
269        Qty::from_units(rounded)
270    }
271
272    pub fn add(&mut self, is_sell: bool, qty: Qty) {
273        if is_sell {
274            self.sell_count += 1;
275            self.sell_sum = Qty::from_units(self.sell_sum.units.saturating_add(qty.units));
276        } else {
277            self.buy_count += 1;
278            self.buy_sum = Qty::from_units(self.buy_sum.units.saturating_add(qty.units));
279        }
280    }
281
282    pub fn values_for(&self, ratio_kind: StackedBarRatio) -> Option<HistAggValues> {
283        match ratio_kind {
284            StackedBarRatio::Count => {
285                let buy = self.buy_count;
286                let sell = self.sell_count;
287                let total = buy.saturating_add(sell);
288
289                if total == 0 {
290                    return None;
291                }
292
293                Some(HistAggValues::Count { buy, sell })
294            }
295            StackedBarRatio::Volume => {
296                let buy = self.buy_sum.to_f32_lossy();
297                let sell = self.sell_sum.to_f32_lossy();
298                let total = buy + sell;
299
300                if total <= 0.0 {
301                    return None;
302                }
303
304                Some(HistAggValues::Qty {
305                    buy: self.buy_sum,
306                    sell: self.sell_sum,
307                })
308            }
309            StackedBarRatio::AverageSize => {
310                let buy_avg = Self::average_qty(self.buy_sum, self.buy_count);
311                let sell_avg = Self::average_qty(self.sell_sum, self.sell_count);
312
313                let denom = buy_avg.units.saturating_add(sell_avg.units);
314                if denom <= 0 {
315                    return None;
316                }
317
318                Some(HistAggValues::Qty {
319                    buy: buy_avg,
320                    sell: sell_avg,
321                })
322            }
323        }
324    }
325}
flowsurface_data/panel/timesales.rs

flowsurface_data/panel/
timesales.rs
