egui_charts/services/detectors/
stops.rs

1//! Same-side execution bursts and bars that sweep through diagonal imbalances.
2
3use std::collections::{BTreeMap, VecDeque};
4
5use chrono::{DateTime, TimeDelta, Utc};
6
7use super::{DetectionKind, Detector};
8use crate::data::{Footprint, meets_threshold};
9use crate::model::{BarClock, Direction, OrderSide, Signal, SignalId, SignalShape, Tick};
10use crate::services::SignalEvent;
11
12/// A burst of at least five same-side fills, at most 20 ms apart, travelling at least three
13/// ticks without reversing. If any fill identifies its aggressor, at least three distinct
14/// aggressors are required. Unknown-side fills are excluded.
15#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
16pub struct StopRunRule;
17
18/// A closed bar with directional delta of at least 25%, three diagonal imbalances at 3:1,
19/// a body at least 60% of its range, and volume at least 1.5 times the previous ten traded bars.
20/// Unknown-side fills never count. The latest subscribed time interval supplies the bars.
21/// The first retained bar is left out because its opening executions may be missing.
22#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
23pub struct StopSweepRule;
24
25#[derive(Default)]
26pub(super) struct StopRuns {
27    burst: Option<Burst>,
28    next: u64,
29}
30
31struct Burst {
32    first: DateTime<Utc>,
33    last: DateTime<Utc>,
34    side: OrderSide,
35    open: i64,
36    price: i64,
37    fills: u64,
38    size: f64,
39    // Only three distinct ids are needed, even for a session-long burst.
40    ids: Vec<u64>,
41    number: Option<u64>,
42}
43
44impl StopRuns {
45    pub(super) fn trade(
46        &mut self,
47        fill: &Tick,
48        tick: f64,
49        enabled: bool,
50        out: &mut Vec<SignalEvent>,
51    ) {
52        let Some(side) = fill.side.filter(|_| fill.size > 0.0) else {
53            return;
54        };
55        let price = (fill.price / tick).round() as i64;
56        if self.burst.as_ref().is_none_or(|burst| {
57            let gap = fill.ts - burst.last;
58            side != burst.side
59                || gap < TimeDelta::zero()
60                || gap > TimeDelta::milliseconds(20)
61                || match side {
62                    OrderSide::Buy => price < burst.price,
63                    OrderSide::Sell => price > burst.price,
64                }
65        }) {
66            self.burst = Some(Burst {
67                first: fill.ts,
68                last: fill.ts,
69                side,
70                open: price,
71                price,
72                fills: 0,
73                size: 0.0,
74                ids: Vec::new(),
75                number: None,
76            });
77        }
78        let burst = self.burst.as_mut().unwrap();
79        burst.last = fill.ts;
80        burst.price = price;
81        burst.fills = burst.fills.saturating_add(1);
82        burst.size = (burst.size + fill.size).min(f64::MAX);
83        if let Some(id) = fill.aggressor_order_id
84            && burst.ids.len() < 3
85            && !burst.ids.contains(&id)
86        {
87            burst.ids.push(id);
88        }
89        let ticks = burst.price.abs_diff(burst.open);
90        if !enabled
91            || burst.fills < 5
92            || ticks < 3
93            || (!burst.ids.is_empty() && burst.ids.len() < 3)
94        {
95            return;
96        }
97        let number = *burst.number.get_or_insert_with(|| {
98            self.next += 1;
99            self.next
100        });
101        out.push(SignalEvent::Upsert(Signal {
102            id: SignalId(format!("stop-run-{number}")),
103            shape: SignalShape::Marker {
104                time: fill.ts,
105                price: Some(fill.price),
106                direction: direction(side),
107            },
108            category: DetectionKind::StopRun.category().into(),
109            label: Some("Stop run".into()),
110            tooltip: Some(format!(
111                "{} fills, {ticks} ticks, size {}, {} ms",
112                burst.fills,
113                burst.size,
114                (burst.last - burst.first).num_milliseconds()
115            )),
116        }));
117    }
118}
119
120pub(super) fn direction(side: OrderSide) -> Direction {
121    match side {
122        OrderSide::Buy => Direction::Up,
123        OrderSide::Sell => Direction::Down,
124    }
125}
126
127pub(super) struct StopSweeps {
128    tick: f64,
129    trades: VecDeque<Tick>,
130    clock: Option<BarClock>,
131    footprint: Option<Footprint>,
132    bars: BTreeMap<DateTime<Utc>, Prices>,
133    closed: Option<DateTime<Utc>>,
134    first: Option<DateTime<Utc>>,
135    volumes: VecDeque<f64>,
136    last_time: Option<DateTime<Utc>>,
137}
138
139struct Prices {
140    open: f64,
141    high: f64,
142    low: f64,
143    close: f64,
144}
145
146impl StopSweeps {
147    pub(super) fn new(tick: f64) -> Self {
148        Self {
149            tick,
150            trades: VecDeque::new(),
151            clock: None,
152            footprint: None,
153            bars: BTreeMap::new(),
154            closed: None,
155            first: None,
156            volumes: VecDeque::new(),
157            last_time: None,
158        }
159    }
160
161    pub(super) fn set_clock(
162        &mut self,
163        clock: Option<BarClock>,
164        settings: &Detector<StopSweepRule>,
165        out: &mut Vec<SignalEvent>,
166    ) {
167        self.clock = clock;
168        self.footprint = self.clock.clone().map(|clock| {
169            let mut footprint = Footprint::new(clock, self.tick, 1).unwrap();
170            footprint.set_budget(12, 100_000);
171            footprint
172        });
173        self.bars.clear();
174        self.closed = None;
175        self.first = None;
176        self.volumes.clear();
177        let trades = std::mem::take(&mut self.trades);
178        for trade in &trades {
179            self.aggregate(trade, settings.enabled, out);
180        }
181        self.trades = trades;
182        if let Some(time) = self.last_time {
183            self.advance(time, settings.enabled, out);
184        }
185    }
186
187    pub(super) fn trade(
188        &mut self,
189        fill: &Tick,
190        settings: &Detector<StopSweepRule>,
191        out: &mut Vec<SignalEvent>,
192    ) {
193        if fill.side.is_none()
194            || fill.size <= 0.0
195            || self.trades.back().is_some_and(|last| fill.ts < last.ts)
196        {
197            return;
198        }
199        self.trades.push_back(fill.clone());
200        if self.trades.len() > 100_000
201            && let Some(old) = self.trades.pop_front()
202            && let Some(clock) = &self.clock
203        {
204            let cut = clock.bar_open(old.ts);
205            if self.first.is_none_or(|first| cut > first) {
206                self.first = Some(cut);
207                if cut == clock.bar_open(fill.ts) {
208                    let mut footprint = Footprint::new(clock.clone(), self.tick, 1).unwrap();
209                    footprint.set_budget(12, 100_000);
210                    self.footprint = Some(footprint);
211                    self.bars.clear();
212                    self.volumes.clear();
213                }
214            }
215        }
216        self.aggregate(fill, settings.enabled, out);
217    }
218
219    fn aggregate(&mut self, fill: &Tick, enabled: bool, out: &mut Vec<SignalEvent>) {
220        self.advance(fill.ts, enabled, out);
221        let (Some(clock), Some(footprint)) = (&self.clock, &mut self.footprint) else {
222            return;
223        };
224        let open = clock.bar_open(fill.ts);
225        let first = *self.first.get_or_insert(open);
226        if open <= first || !matches!(footprint.push(fill), Ok(true)) {
227            return;
228        }
229        let bar = self.bars.entry(open).or_insert(Prices {
230            open: fill.price,
231            high: fill.price,
232            low: fill.price,
233            close: fill.price,
234        });
235        bar.high = bar.high.max(fill.price);
236        bar.low = bar.low.min(fill.price);
237        bar.close = fill.price;
238        while self.bars.len() > 12 {
239            self.bars.pop_first();
240        }
241    }
242
243    pub(super) fn advance(
244        &mut self,
245        time: DateTime<Utc>,
246        enabled: bool,
247        out: &mut Vec<SignalEvent>,
248    ) {
249        self.last_time = Some(self.last_time.map_or(time, |last| last.max(time)));
250        let (Some(clock), Some(footprint)) = (&self.clock, &self.footprint) else {
251            return;
252        };
253        let open = clock.bar_open(time);
254        for (&at, prices) in self.bars.range(..open) {
255            if self.closed.is_some_and(|closed| at <= closed) {
256                continue;
257            }
258            self.closed = Some(at);
259            if self.first.is_some_and(|first| at <= first) {
260                continue;
261            }
262            let Some(bar) = footprint.bar_at(at) else {
263                continue;
264            };
265            let volume = bar.volume();
266            let buy = prices.close > prices.open;
267            let delta = if buy { bar.delta() } else { -bar.delta() };
268            let count = bar
269                .levels
270                .keys()
271                .filter(|&&level| {
272                    let (bid, ask) = bar.imbalance(level, 3.0, true, 0.0);
273                    if buy { ask } else { bid }
274                })
275                .count();
276            if enabled
277                && prices.high > prices.low
278                && meets_threshold(delta, volume * 0.25)
279                && meets_threshold(
280                    (prices.close - prices.open).abs(),
281                    (prices.high - prices.low) * 0.6,
282                )
283                && count >= 3
284                && self.volumes.len() == 10
285                && meets_threshold(
286                    volume,
287                    1.5 * self.volumes.iter().map(|volume| volume / 10.0).sum::<f64>(),
288                )
289            {
290                out.push(SignalEvent::Upsert(Signal {
291                    id: SignalId(format!("stop-sweep-{}", at.timestamp_millis())),
292                    shape: SignalShape::Marker {
293                        time: at,
294                        price: Some(prices.close),
295                        direction: if buy { Direction::Up } else { Direction::Down },
296                    },
297                    category: DetectionKind::StopSweep.category().into(),
298                    label: Some("Stop sweep".into()),
299                    tooltip: Some(format!(
300                        "Volume {volume}, delta {}, {count} diagonal imbalances",
301                        bar.delta()
302                    )),
303                }));
304            }
305            self.volumes.push_back(volume);
306            if self.volumes.len() > 10 {
307                self.volumes.pop_front();
308            }
309        }
310    }
311}
egui_charts/services/detectors/stops.rs

egui_charts/services/detectors/
stops.rs
