egui_charts/services/detectors/
alerts.rs

1//! Feed-time counts and live notifications, independent of retained chart signals.
2
3use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
4
5use chrono::{DateTime, TimeDelta, Utc};
6
7use super::{DetectionKind, DetectorSettings, OrderFlowDetectors, RETAINED, Tracked};
8use crate::model::SignalId;
9use crate::services::{Events, Notification, Notifications, SignalEvent};
10
11/// Running detections, including loaded history and excluding updates of an existing signal.
12#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
13pub struct DetectionCounts {
14    /// Detections since the symbol was tracked or reset, including evicted chart signals.
15    pub total: u64,
16    /// Detections in `(now - 5 minutes, now]` by feed time.
17    pub last_five_minutes: u64,
18    /// Detections in `(now - 30 seconds, now]` by feed time.
19    pub last_thirty_seconds: u64,
20}
21
22/// The feed-time window of a count alert.
23#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
24pub enum CountWindow {
25    /// Since tracking or reset.
26    Total,
27    /// The preceding five minutes, excluding the opening boundary.
28    FiveMinutes,
29    /// The preceding thirty seconds, excluding the opening boundary.
30    #[default]
31    ThirtySeconds,
32}
33
34impl CountWindow {
35    /// The count of this window.
36    pub fn count(self, counts: DetectionCounts) -> u64 {
37        match self {
38            Self::Total => counts.total,
39            Self::FiveMinutes => counts.last_five_minutes,
40            Self::ThirtySeconds => counts.last_thirty_seconds,
41        }
42    }
43
44    /// The window label, also a translation key for user interfaces.
45    pub fn label(self) -> &'static str {
46        match self {
47            Self::Total => "Since tracking began",
48            Self::FiveMinutes => "Last 5 minutes",
49            Self::ThirtySeconds => "Last 30 seconds",
50        }
51    }
52}
53
54/// Fires once when its window reaches `minimum`, rearming only below it.
55#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
56#[non_exhaustive]
57pub struct CountAlert {
58    /// Which running count to compare.
59    pub window: CountWindow,
60    /// Positive minimum, validated by [`OrderFlowDetectors::set_settings`].
61    pub minimum: u64,
62    /// Whether the host should play a sound for this count alert.
63    pub sound: bool,
64}
65
66impl Default for CountAlert {
67    fn default() -> Self {
68        Self {
69            window: CountWindow::default(),
70            minimum: 1,
71            sound: false,
72        }
73    }
74}
75
76/// A live detection or count alert for a panel to show and the host to sound.
77#[derive(Clone, Debug, PartialEq)]
78#[non_exhaustive]
79pub struct DetectionAlert {
80    /// The detector category.
81    pub kind: DetectionKind,
82    /// The same entry delivered through [`Notifications::watch`].
83    pub notification: Notification,
84    /// The window and reached count for a count alert; absent for an individual detection.
85    pub count: Option<(CountWindow, u64)>,
86    /// Whether the host should play this kind's sound.
87    pub sound: bool,
88}
89
90#[derive(Default)]
91pub(super) struct Delivery {
92    alerts: VecDeque<DetectionAlert>,
93    log: VecDeque<Notification>,
94    watchers: Vec<Events<Notification>>,
95}
96
97impl Delivery {
98    fn push(&mut self, alert: DetectionAlert) {
99        self.watchers.retain(Events::is_active);
100        for events in &self.watchers {
101            events.push(alert.notification.clone());
102        }
103        self.log.push_back(alert.notification.clone());
104        self.alerts.push_back(alert);
105        if self.log.len() > RETAINED {
106            self.log.pop_front();
107        }
108        if self.alerts.len() > RETAINED {
109            self.alerts.pop_front();
110        }
111    }
112}
113
114impl OrderFlowDetectors {
115    /// Counts of `kind` on `symbol` at the current feed time, or zero for an untracked symbol.
116    /// Windows retain five minutes relative to the latest supplied feed time; queries before
117    /// that retained interval cannot restore expired detections. Wall-clock time is not used.
118    pub fn counts(&self, symbol: &str, kind: DetectionKind, now: DateTime<Utc>) -> DetectionCounts {
119        self.lock()
120            .symbols
121            .get(symbol)
122            .map(|tracked| tracked.counting.counts(kind, now))
123            .unwrap_or_default()
124    }
125
126    /// Drains pending live alerts once, oldest first. At most the latest 5,000 are retained
127    /// between calls; draining does not remove entries from the notification log.
128    pub fn take_alerts(&self) -> Vec<DetectionAlert> {
129        self.lock().delivery.alerts.drain(..).collect()
130    }
131}
132
133impl Notifications for OrderFlowDetectors {
134    fn watch(&self, events: Events<Notification>) {
135        let mut state = self.lock();
136        for notification in &state.delivery.log {
137            events.push(notification.clone());
138        }
139        state.delivery.watchers.retain(Events::is_active);
140        state.delivery.watchers.push(events);
141    }
142}
143
144#[derive(Default)]
145pub(super) struct Counting {
146    kinds: HashMap<DetectionKind, Counter>,
147    now: Option<DateTime<Utc>>,
148}
149
150#[derive(Default)]
151struct Counter {
152    total: u64,
153    times: BTreeMap<DateTime<Utc>, u64>,
154    last_episode: u64,
155    rule: Option<(CountWindow, u64)>,
156    above: bool,
157}
158
159impl Counting {
160    fn counts(&self, kind: DetectionKind, now: DateTime<Utc>) -> DetectionCounts {
161        let Some(counter) = self.kinds.get(&kind) else {
162            return DetectionCounts::default();
163        };
164        let mut counts = DetectionCounts {
165            total: counter.total,
166            ..Default::default()
167        };
168        for (&time, &number) in &counter.times {
169            let age = now.signed_duration_since(time);
170            if age >= TimeDelta::zero() && age < TimeDelta::minutes(5) {
171                counts.last_five_minutes = counts.last_five_minutes.saturating_add(number);
172                if age < TimeDelta::seconds(30) {
173                    counts.last_thirty_seconds = counts.last_thirty_seconds.saturating_add(number);
174                }
175            }
176        }
177        counts
178    }
179
180    pub(super) fn advance(
181        &mut self,
182        time: DateTime<Utc>,
183        settings: &DetectorSettings,
184        live: bool,
185        symbol: &str,
186        delivery: &mut Delivery,
187    ) {
188        let now = self.now.map_or(time, |last| last.max(time));
189        self.now = Some(now);
190        for counter in self.kinds.values_mut() {
191            while counter
192                .times
193                .first_key_value()
194                .is_some_and(|(&time, _)| now.signed_duration_since(time) >= TimeDelta::minutes(5))
195            {
196                counter.times.pop_first();
197            }
198        }
199        for kind in [DetectionKind::Iceberg, DetectionKind::StopRun] {
200            self.check(kind, now, settings, live, symbol, delivery);
201        }
202    }
203
204    fn check(
205        &mut self,
206        kind: DetectionKind,
207        now: DateTime<Utc>,
208        settings: &DetectorSettings,
209        live: bool,
210        symbol: &str,
211        delivery: &mut Delivery,
212    ) {
213        let rule = match kind {
214            DetectionKind::Iceberg => settings.iceberg_count_alert,
215            DetectionKind::StopRun => settings.stop_run_count_alert,
216            _ => return,
217        };
218        let count = rule.map(|rule| rule.window.count(self.counts(kind, now)));
219        let counter = self.kinds.entry(kind).or_default();
220        let edge = rule.map(|rule| (rule.window, rule.minimum));
221        if counter.rule != edge {
222            counter.above = false;
223            counter.rule = edge;
224        }
225        let above = rule
226            .zip(count)
227            .is_some_and(|(rule, count)| count >= rule.minimum);
228        if above && !counter.above && live {
229            let rule = rule.unwrap();
230            let count = count.unwrap();
231            let mut notification = Notification::new(now, kind.category());
232            notification.body = format!("{}: {count}", rule.window.label());
233            notification.symbol = Some(symbol.into());
234            delivery.push(DetectionAlert {
235                kind,
236                notification,
237                count: Some((rule.window, count)),
238                sound: rule.sound,
239            });
240        }
241        counter.above = above;
242    }
243}
244
245impl Tracked {
246    pub(super) fn record(
247        &mut self,
248        events: &[SignalEvent],
249        time: DateTime<Utc>,
250        live: bool,
251        settings: &DetectorSettings,
252        delivery: &mut Delivery,
253        added: &mut HashSet<SignalId>,
254    ) {
255        for event in events {
256            let SignalEvent::Upsert(signal) = event else {
257                continue;
258            };
259            if self.detections.iter().any(|kept| kept.id == signal.id)
260                || !added.insert(signal.id.clone())
261            {
262                continue;
263            }
264            let Some(kind) = DetectionKind::ALL
265                .into_iter()
266                .find(|kind| kind.category() == signal.category)
267            else {
268                continue;
269            };
270            let counter = self.counting.kinds.entry(kind).or_default();
271            // These episodes can outlive chart retention; their numbers only increase.
272            if matches!(
273                kind,
274                DetectionKind::BigTrade
275                    | DetectionKind::BigTradeZone
276                    | DetectionKind::StopRun
277                    | DetectionKind::Iceberg
278            ) {
279                let number = signal
280                    .id
281                    .0
282                    .rsplit('-')
283                    .next()
284                    .unwrap()
285                    .parse::<u64>()
286                    .expect("numbered detector episode");
287                if number <= counter.last_episode {
288                    continue;
289                }
290                counter.last_episode = number;
291            }
292            counter.total = counter.total.saturating_add(1);
293            let number = counter.times.entry(time).or_default();
294            *number = number.saturating_add(1);
295            let (alert, sound) = match kind {
296                DetectionKind::BigTrade => (settings.big_trades.alert, settings.big_trades.sound),
297                // A zone is the price range of a trade that has already alerted.
298                DetectionKind::BigTradeZone => (false, false),
299                DetectionKind::StopRun => (settings.stop_runs.alert, settings.stop_runs.sound),
300                DetectionKind::StopSweep => {
301                    (settings.stop_sweeps.alert, settings.stop_sweeps.sound)
302                }
303                DetectionKind::Iceberg => (settings.icebergs.alert, settings.icebergs.sound),
304                DetectionKind::Absorption => (settings.absorption.alert, settings.absorption.sound),
305                _ => (settings.liquidity.alert, settings.liquidity.sound),
306            };
307            if live && alert {
308                let mut notification = Notification::new(time, kind.category());
309                notification.body = signal
310                    .tooltip
311                    .as_ref()
312                    .or(signal.label.as_ref())
313                    .cloned()
314                    .unwrap_or_else(|| kind.category().into());
315                notification.symbol = Some(self.instrument.symbol.clone());
316                delivery.push(DetectionAlert {
317                    kind,
318                    notification,
319                    count: None,
320                    sound,
321                });
322            }
323            self.counting.check(
324                kind,
325                self.counting.now.unwrap_or(time),
326                settings,
327                live,
328                &self.instrument.symbol,
329                delivery,
330            );
331        }
332    }
333}
egui_charts/services/detectors/alerts.rs

egui_charts/services/detectors/
alerts.rs
