finance_query/streaming/
alerts.rs

1//! Threshold-triggered alerting over a price subscription.
2//!
3//! Consumers that only care about specific crossings would otherwise receive
4//! every tick and filter client-side. [`AlertEvaluator`] moves that predicate
5//! next to the stream; [`AlertStream`] wraps it as a `Stream` adapter so it
6//! composes with any `Stream<Item = PriceUpdate>` (including a server-side
7//! shared hub stream) rather than being welded to one transport.
8
9use std::collections::{HashMap, HashSet, VecDeque};
10use std::pin::Pin;
11use std::str::FromStr;
12use std::task::{Context, Poll};
13
14use futures::stream::Stream;
15use serde::{Deserialize, Serialize};
16
17use super::pricing::PriceUpdate;
18use crate::error::FinanceError;
19
20/// A predicate over incoming price ticks.
21#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
22#[serde(rename_all = "camelCase")]
23#[non_exhaustive]
24pub enum AlertCondition {
25    /// Price moved from at-or-below the threshold to above it.
26    CrossesAbove(f64),
27    /// Price moved from at-or-above the threshold to below it.
28    CrossesBelow(f64),
29    /// Price is above the threshold (fires on the first matching tick).
30    PriceAbove(f64),
31    /// Price is below the threshold (fires on the first matching tick).
32    PriceBelow(f64),
33    /// Percent change from the previous close is at or above the threshold.
34    PercentChangeAbove(f64),
35    /// Percent change from the previous close is at or below the threshold.
36    PercentChangeBelow(f64),
37    /// Day volume is at or above the threshold.
38    VolumeAbove(i64),
39}
40
41/// Which predicate an [`AlertCondition`] applies, without its threshold.
42///
43/// Deliberately **exhaustive** (no `#[non_exhaustive]`): transports that
44/// project a condition onto a flat `kind` + `value` pair must fail to compile
45/// when a predicate is added, rather than silently degrading it.
46#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
47#[serde(rename_all = "camelCase")]
48pub enum AlertConditionKind {
49    /// See [`AlertCondition::CrossesAbove`].
50    CrossesAbove,
51    /// See [`AlertCondition::CrossesBelow`].
52    CrossesBelow,
53    /// See [`AlertCondition::PriceAbove`].
54    PriceAbove,
55    /// See [`AlertCondition::PriceBelow`].
56    PriceBelow,
57    /// See [`AlertCondition::PercentChangeAbove`].
58    PercentChangeAbove,
59    /// See [`AlertCondition::PercentChangeBelow`].
60    PercentChangeBelow,
61    /// See [`AlertCondition::VolumeAbove`].
62    VolumeAbove,
63}
64
65impl AlertConditionKind {
66    /// Pair this predicate with a threshold to get a usable condition.
67    pub fn with_value(self, value: f64) -> AlertCondition {
68        match self {
69            Self::CrossesAbove => AlertCondition::CrossesAbove(value),
70            Self::CrossesBelow => AlertCondition::CrossesBelow(value),
71            Self::PriceAbove => AlertCondition::PriceAbove(value),
72            Self::PriceBelow => AlertCondition::PriceBelow(value),
73            Self::PercentChangeAbove => AlertCondition::PercentChangeAbove(value),
74            Self::PercentChangeBelow => AlertCondition::PercentChangeBelow(value),
75            Self::VolumeAbove => AlertCondition::VolumeAbove(value as i64),
76        }
77    }
78
79    /// Wire name of this predicate (`"crossesAbove"`, …).
80    pub fn as_str(self) -> &'static str {
81        match self {
82            Self::CrossesAbove => "crossesAbove",
83            Self::CrossesBelow => "crossesBelow",
84            Self::PriceAbove => "priceAbove",
85            Self::PriceBelow => "priceBelow",
86            Self::PercentChangeAbove => "percentChangeAbove",
87            Self::PercentChangeBelow => "percentChangeBelow",
88            Self::VolumeAbove => "volumeAbove",
89        }
90    }
91}
92
93impl std::fmt::Display for AlertConditionKind {
94    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95        f.write_str(self.as_str())
96    }
97}
98
99impl FromStr for AlertConditionKind {
100    type Err = FinanceError;
101
102    fn from_str(s: &str) -> Result<Self, Self::Err> {
103        match s {
104            "crossesAbove" => Ok(Self::CrossesAbove),
105            "crossesBelow" => Ok(Self::CrossesBelow),
106            "priceAbove" => Ok(Self::PriceAbove),
107            "priceBelow" => Ok(Self::PriceBelow),
108            "percentChangeAbove" => Ok(Self::PercentChangeAbove),
109            "percentChangeBelow" => Ok(Self::PercentChangeBelow),
110            "volumeAbove" => Ok(Self::VolumeAbove),
111            other => Err(FinanceError::InvalidParameter {
112                param: "condition".to_string(),
113                reason: format!("unknown alert condition: {other}"),
114            }),
115        }
116    }
117}
118
119impl AlertCondition {
120    /// Which predicate this condition applies.
121    pub fn kind(&self) -> AlertConditionKind {
122        match *self {
123            Self::CrossesAbove(_) => AlertConditionKind::CrossesAbove,
124            Self::CrossesBelow(_) => AlertConditionKind::CrossesBelow,
125            Self::PriceAbove(_) => AlertConditionKind::PriceAbove,
126            Self::PriceBelow(_) => AlertConditionKind::PriceBelow,
127            Self::PercentChangeAbove(_) => AlertConditionKind::PercentChangeAbove,
128            Self::PercentChangeBelow(_) => AlertConditionKind::PercentChangeBelow,
129            Self::VolumeAbove(_) => AlertConditionKind::VolumeAbove,
130        }
131    }
132
133    /// Threshold this condition compares against.
134    pub fn threshold(&self) -> f64 {
135        match *self {
136            Self::CrossesAbove(t)
137            | Self::CrossesBelow(t)
138            | Self::PriceAbove(t)
139            | Self::PriceBelow(t)
140            | Self::PercentChangeAbove(t)
141            | Self::PercentChangeBelow(t) => t,
142            Self::VolumeAbove(t) => t as f64,
143        }
144    }
145
146    /// Whether the condition holds for this tick.
147    ///
148    /// `previous` is the last price seen for the symbol; crossing conditions
149    /// need it and never fire on the first tick of a symbol.
150    fn holds(&self, update: &PriceUpdate, previous: Option<f32>) -> bool {
151        let price = update.price as f64;
152        match *self {
153            Self::CrossesAbove(t) => previous.is_some_and(|p| (p as f64) <= t) && price > t,
154            Self::CrossesBelow(t) => previous.is_some_and(|p| (p as f64) >= t) && price < t,
155            Self::PriceAbove(t) => price > t,
156            Self::PriceBelow(t) => price < t,
157            Self::PercentChangeAbove(t) => update.change_percent as f64 >= t,
158            Self::PercentChangeBelow(t) => update.change_percent as f64 <= t,
159            Self::VolumeAbove(t) => update.day_volume >= t,
160        }
161    }
162}
163
164/// A symbol paired with the condition that should trigger an alert.
165#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
166#[serde(rename_all = "camelCase")]
167#[non_exhaustive]
168pub struct AlertRule {
169    /// Symbol this rule watches (matched against `PriceUpdate::id`).
170    pub symbol: String,
171    /// Predicate to evaluate.
172    pub condition: AlertCondition,
173    /// Fire repeatedly (re-arming whenever the condition stops holding)
174    /// instead of once.
175    pub repeat: bool,
176}
177
178impl AlertRule {
179    /// A one-shot rule: fires the first time the condition holds.
180    pub fn new(symbol: impl Into<String>, condition: AlertCondition) -> Self {
181        Self {
182            symbol: symbol.into(),
183            condition,
184            repeat: false,
185        }
186    }
187
188    /// Fire every time the condition becomes true again.
189    pub fn repeating(mut self) -> Self {
190        self.repeat = true;
191        self
192    }
193}
194
195/// A fired alert, carrying the tick that triggered it.
196#[derive(Clone, Debug, Serialize, Deserialize)]
197#[serde(rename_all = "camelCase")]
198#[non_exhaustive]
199pub struct AlertEvent {
200    /// Symbol that triggered.
201    pub symbol: String,
202    /// Condition that fired.
203    pub condition: AlertCondition,
204    /// Price on the triggering tick.
205    pub price: f32,
206    /// Last price seen before this tick, if any.
207    pub previous_price: Option<f32>,
208    /// Percent change carried by the triggering tick.
209    pub change_percent: f32,
210    /// Tick timestamp (milliseconds).
211    pub time: i64,
212    /// The full triggering tick.
213    pub update: PriceUpdate,
214}
215
216/// Per-rule firing state.
217struct RuleState {
218    rule: AlertRule,
219    /// `false` after firing; a repeating rule re-arms when the condition
220    /// stops holding, a one-shot rule never does.
221    armed: bool,
222}
223
224/// Evaluates [`AlertRule`]s against a price feed, tracking the per-symbol
225/// history that crossing conditions need.
226///
227/// Exposed so consumers that already own a price stream (the server's shared
228/// hub, for instance) can apply alerts without re-wrapping the stream.
229pub struct AlertEvaluator {
230    rules: Vec<RuleState>,
231    /// Symbols any rule watches — a shared feed carries far more than these,
232    /// and untracked ids must not accumulate history.
233    watched: HashSet<String>,
234    last_price: HashMap<String, f32>,
235}
236
237impl AlertEvaluator {
238    /// Build an evaluator from a rule set.
239    pub fn new(rules: impl IntoIterator<Item = AlertRule>) -> Self {
240        let rules: Vec<RuleState> = rules
241            .into_iter()
242            .map(|rule| RuleState { rule, armed: true })
243            .collect();
244        Self {
245            watched: rules.iter().map(|s| s.rule.symbol.clone()).collect(),
246            rules,
247            last_price: HashMap::new(),
248        }
249    }
250
251    /// Distinct symbols referenced by the rule set, in first-seen order.
252    pub fn symbols(&self) -> Vec<String> {
253        let mut symbols: Vec<String> = Vec::new();
254        for state in &self.rules {
255            if !symbols.contains(&state.rule.symbol) {
256                symbols.push(state.rule.symbol.clone());
257            }
258        }
259        symbols
260    }
261
262    /// `true` once every rule has fired and none can fire again.
263    pub fn is_exhausted(&self) -> bool {
264        self.rules
265            .iter()
266            .all(|state| !state.armed && !state.rule.repeat)
267    }
268
269    /// Feed one tick, returning the alerts it triggered.
270    pub fn evaluate(&mut self, update: &PriceUpdate) -> Vec<AlertEvent> {
271        if !self.watched.contains(&update.id) {
272            return Vec::new();
273        }
274        let previous = self.last_price.get(&update.id).copied();
275        let mut fired = Vec::new();
276
277        for state in self.rules.iter_mut() {
278            if state.rule.symbol != update.id {
279                continue;
280            }
281            let holds = state.rule.condition.holds(update, previous);
282            if holds && state.armed {
283                state.armed = false;
284                fired.push(AlertEvent {
285                    symbol: update.id.clone(),
286                    condition: state.rule.condition,
287                    price: update.price,
288                    previous_price: previous,
289                    change_percent: update.change_percent,
290                    time: update.time,
291                    update: update.clone(),
292                });
293            } else if !holds && state.rule.repeat {
294                state.armed = true;
295            }
296        }
297
298        // Heartbeats and other priceless ticks must not poison the crossing
299        // history with a zero.
300        if update.price != 0.0 {
301            match self.last_price.get_mut(&update.id) {
302                Some(last) => *last = update.price,
303                None => {
304                    self.last_price.insert(update.id.clone(), update.price);
305                }
306            }
307        }
308        fired
309    }
310}
311
312/// A `Stream<Item = AlertEvent>` over any price stream.
313pub struct AlertStream<S> {
314    inner: S,
315    evaluator: AlertEvaluator,
316    pending: VecDeque<AlertEvent>,
317}
318
319impl<S> AlertStream<S> {
320    /// Wrap `inner`, evaluating `rules` against every tick it yields.
321    pub fn new(inner: S, rules: impl IntoIterator<Item = AlertRule>) -> Self {
322        Self {
323            inner,
324            evaluator: AlertEvaluator::new(rules),
325            pending: VecDeque::new(),
326        }
327    }
328}
329
330impl<S> Stream for AlertStream<S>
331where
332    S: Stream<Item = PriceUpdate> + Unpin,
333{
334    type Item = AlertEvent;
335
336    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
337        let this = self.get_mut();
338        loop {
339            if let Some(event) = this.pending.pop_front() {
340                return Poll::Ready(Some(event));
341            }
342            match Pin::new(&mut this.inner).poll_next(cx) {
343                Poll::Ready(Some(update)) => this.pending.extend(this.evaluator.evaluate(&update)),
344                Poll::Ready(None) => return Poll::Ready(None),
345                Poll::Pending => return Poll::Pending,
346            }
347        }
348    }
349}
350
351/// Adds [`alerts`](AlertExt::alerts) to every price stream.
352///
353/// # Example
354///
355/// ```no_run
356/// use finance_query::streaming::{AlertCondition, AlertExt, AlertRule, PriceStream};
357/// use futures::StreamExt;
358///
359/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
360/// let mut alerts = PriceStream::subscribe(["AAPL"])
361///     .await?
362///     .alerts([AlertRule::new("AAPL", AlertCondition::CrossesAbove(200.0))]);
363///
364/// while let Some(alert) = alerts.next().await {
365///     println!("{} crossed at {}", alert.symbol, alert.price);
366/// }
367/// # Ok(())
368/// # }
369/// ```
370pub trait AlertExt: Stream<Item = PriceUpdate> + Sized + Unpin {
371    /// Yield only the ticks that trigger one of `rules`.
372    fn alerts(self, rules: impl IntoIterator<Item = AlertRule>) -> AlertStream<Self> {
373        AlertStream::new(self, rules)
374    }
375}
376
377impl<S> AlertExt for S where S: Stream<Item = PriceUpdate> + Sized + Unpin {}
378
379#[cfg(test)]
380mod tests {
381    use super::*;
382    use futures::StreamExt;
383
384    fn tick(symbol: &str, price: f32) -> PriceUpdate {
385        PriceUpdate {
386            id: symbol.to_string(),
387            price,
388            ..Default::default()
389        }
390    }
391
392    #[test]
393    fn crossing_needs_a_previous_price() {
394        let mut evaluator =
395            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::CrossesAbove(150.0))]);
396
397        // First tick is already above the threshold but has no history.
398        assert!(evaluator.evaluate(&tick("AAPL", 155.0)).is_empty());
399    }
400
401    #[test]
402    fn crossing_fires_once_on_the_upward_move() {
403        let mut evaluator =
404            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::CrossesAbove(150.0))]);
405
406        assert!(evaluator.evaluate(&tick("AAPL", 149.0)).is_empty());
407        let fired = evaluator.evaluate(&tick("AAPL", 151.0));
408        assert_eq!(fired.len(), 1);
409        assert_eq!(fired[0].previous_price, Some(149.0));
410
411        // One-shot: no repeat while it stays above, and none on a re-cross.
412        assert!(evaluator.evaluate(&tick("AAPL", 152.0)).is_empty());
413        assert!(evaluator.evaluate(&tick("AAPL", 148.0)).is_empty());
414        assert!(evaluator.evaluate(&tick("AAPL", 153.0)).is_empty());
415        assert!(evaluator.is_exhausted());
416    }
417
418    #[test]
419    fn repeating_rules_rearm_when_the_condition_clears() {
420        let mut evaluator =
421            AlertEvaluator::new([
422                AlertRule::new("AAPL", AlertCondition::CrossesAbove(150.0)).repeating()
423            ]);
424
425        evaluator.evaluate(&tick("AAPL", 149.0));
426        assert_eq!(evaluator.evaluate(&tick("AAPL", 151.0)).len(), 1);
427        assert!(evaluator.evaluate(&tick("AAPL", 152.0)).is_empty());
428        // Falls back below (clearing the condition), then crosses again.
429        assert!(evaluator.evaluate(&tick("AAPL", 148.0)).is_empty());
430        assert_eq!(evaluator.evaluate(&tick("AAPL", 151.0)).len(), 1);
431        assert!(!evaluator.is_exhausted());
432    }
433
434    #[test]
435    fn crossing_below_is_symmetric() {
436        let mut evaluator =
437            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::CrossesBelow(100.0))]);
438        evaluator.evaluate(&tick("AAPL", 101.0));
439        assert_eq!(evaluator.evaluate(&tick("AAPL", 99.0)).len(), 1);
440    }
441
442    #[test]
443    fn level_and_metric_conditions_fire_without_history() {
444        let mut level =
445            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::PriceAbove(10.0))]);
446        assert_eq!(level.evaluate(&tick("AAPL", 11.0)).len(), 1);
447
448        let mut pct = AlertEvaluator::new([AlertRule::new(
449            "AAPL",
450            AlertCondition::PercentChangeAbove(5.0),
451        )]);
452        let mut update = tick("AAPL", 11.0);
453        update.change_percent = 6.0;
454        assert_eq!(pct.evaluate(&update).len(), 1);
455
456        let mut vol =
457            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::VolumeAbove(1_000))]);
458        let mut update = tick("AAPL", 11.0);
459        update.day_volume = 1_500;
460        assert_eq!(vol.evaluate(&update).len(), 1);
461    }
462
463    #[test]
464    fn rules_only_see_their_own_symbol() {
465        let mut evaluator = AlertEvaluator::new([
466            AlertRule::new("AAPL", AlertCondition::PriceAbove(10.0)),
467            AlertRule::new("NVDA", AlertCondition::PriceAbove(10.0)),
468        ]);
469        let fired = evaluator.evaluate(&tick("NVDA", 20.0));
470        assert_eq!(fired.len(), 1);
471        assert_eq!(fired[0].symbol, "NVDA");
472        assert_eq!(evaluator.symbols(), vec!["AAPL", "NVDA"]);
473    }
474
475    #[test]
476    fn priceless_ticks_do_not_become_crossing_history() {
477        let mut evaluator =
478            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::CrossesAbove(150.0))]);
479        evaluator.evaluate(&tick("AAPL", 149.0));
480        // A heartbeat-style tick with no price must not reset the baseline.
481        evaluator.evaluate(&tick("AAPL", 0.0));
482        assert_eq!(evaluator.evaluate(&tick("AAPL", 151.0)).len(), 1);
483    }
484
485    #[test]
486    fn unwatched_symbols_leave_no_trace() {
487        let mut evaluator =
488            AlertEvaluator::new([AlertRule::new("AAPL", AlertCondition::CrossesAbove(150.0))]);
489        // A shared hub carries every symbol any client subscribed to.
490        assert!(evaluator.evaluate(&tick("TSLA", 400.0)).is_empty());
491        assert!(!evaluator.last_price.contains_key("TSLA"));
492    }
493
494    #[test]
495    fn conditions_project_onto_kind_and_threshold() {
496        for condition in [
497            AlertCondition::CrossesAbove(1.5),
498            AlertCondition::CrossesBelow(1.5),
499            AlertCondition::PriceAbove(1.5),
500            AlertCondition::PriceBelow(1.5),
501            AlertCondition::PercentChangeAbove(1.5),
502            AlertCondition::PercentChangeBelow(1.5),
503        ] {
504            let round_tripped = condition.kind().with_value(condition.threshold());
505            assert_eq!(round_tripped, condition);
506        }
507
508        let volume = AlertCondition::VolumeAbove(1_000);
509        assert_eq!(volume.kind(), AlertConditionKind::VolumeAbove);
510        assert_eq!(volume.kind().with_value(volume.threshold()), volume);
511    }
512
513    #[test]
514    fn condition_kinds_round_trip_through_their_wire_names() {
515        for kind in [
516            AlertConditionKind::CrossesAbove,
517            AlertConditionKind::CrossesBelow,
518            AlertConditionKind::PriceAbove,
519            AlertConditionKind::PriceBelow,
520            AlertConditionKind::PercentChangeAbove,
521            AlertConditionKind::PercentChangeBelow,
522            AlertConditionKind::VolumeAbove,
523        ] {
524            assert_eq!(kind.as_str().parse::<AlertConditionKind>().unwrap(), kind);
525        }
526        assert!("wat".parse::<AlertConditionKind>().is_err());
527    }
528
529    #[tokio::test]
530    async fn stream_adapter_yields_only_triggering_ticks() {
531        let updates = futures::stream::iter(vec![
532            tick("AAPL", 149.0),
533            tick("AAPL", 149.5),
534            tick("AAPL", 151.0),
535            tick("AAPL", 152.0),
536        ]);
537
538        let alerts: Vec<AlertEvent> = updates
539            .alerts([AlertRule::new("AAPL", AlertCondition::CrossesAbove(150.0))])
540            .collect()
541            .await;
542
543        assert_eq!(alerts.len(), 1);
544        assert_eq!(alerts[0].price, 151.0);
545        assert_eq!(alerts[0].update.id, "AAPL");
546    }
547}
finance_query/streaming/alerts.rs

finance_query/streaming/
alerts.rs
