1pub mod indicators;
2
3use super::Basis;
4use super::aggr::time::DataPoint;
5use exchange::unit::MinQtySize;
6use exchange::unit::price::{Price, PriceStep};
7use exchange::unit::qty::{Qty, SizeUnit, volume_size_unit};
8use exchange::{Timeframe, UnixMs, adapter::MarketKind, depth::Depth};
9
10use rustc_hash::{FxBuildHasher, FxHashMap};
11use serde::{Deserialize, Serialize};
12use std::{collections::BTreeMap, ops::ControlFlow};
13
14pub const CLEANUP_THRESHOLD: usize = 4800;
15
16#[derive(Debug, Copy, Clone, PartialEq, Deserialize, Serialize)]
17pub struct Config {
18 pub trade_size_filter: f32,
19 pub order_size_filter: f32,
20 pub trade_size_scale: Option<i32>,
21 pub coalescing: Option<CoalesceKind>,
22}
23
24impl Default for Config {
25 fn default() -> Self {
26 Config {
27 trade_size_filter: 0.0,
28 order_size_filter: 0.0,
29 trade_size_scale: Some(100),
30 coalescing: Some(CoalesceKind::Average(0.15)),
31 }
32 }
33}
34
35#[derive(Debug, Clone)]
36pub struct SettingsSnapshot {
37 pub config: Option<Config>,
38 pub indicators: indicators::HeatmapIndicatorConfig,
39}
40
41#[derive(Default)]
42pub struct HeatmapDataPoint {
43 pub grouped_trades: Box<[GroupedTrade]>,
44 pub buy_sell: (Qty, Qty),
45}
46
47impl DataPoint for HeatmapDataPoint {
48 fn add_trade(&mut self, trade: &exchange::Trade, step: PriceStep) {
49 let grouped_price: Price = trade.price.round_to_side_step(trade.is_sell, step);
50
51 match self
52 .grouped_trades
53 .binary_search_by(|probe| probe.compare_with(grouped_price, trade.is_sell))
54 {
55 Ok(index) => self.grouped_trades[index].qty += trade.qty,
56 Err(index) => {
57 let mut trades = self.grouped_trades.to_vec();
58 trades.insert(
59 index,
60 GroupedTrade {
61 is_sell: trade.is_sell,
62 price: grouped_price,
63 qty: trade.qty,
64 },
65 );
66 self.grouped_trades = trades.into_boxed_slice();
67 }
68 }
69
70 if trade.is_sell {
71 self.buy_sell.1 += trade.qty;
72 } else {
73 self.buy_sell.0 += trade.qty;
74 }
75 }
76
77 fn clear_trades(&mut self) {
78 self.grouped_trades = Box::new([]);
79 self.buy_sell = (Qty::default(), Qty::default());
80 }
81}
82
83#[derive(Default, Debug, Clone, Copy, PartialEq)]
84pub struct OrderRun {
85 pub start_time: UnixMs,
86 pub until_time: UnixMs,
87 pub qty: Qty,
88 pub is_bid: bool,
89}
90
91impl OrderRun {
92 pub fn new(start_time: UnixMs, aggr_time: Timeframe, qty: Qty, is_bid: bool) -> Self {
93 OrderRun {
94 start_time,
95 until_time: start_time.saturating_add(aggr_time.to_milliseconds()),
96 qty,
97 is_bid,
98 }
99 }
100
101 pub fn with_range(&self, earliest: UnixMs, latest: UnixMs) -> Option<&OrderRun> {
102 if self.start_time <= latest && self.until_time >= earliest {
103 Some(self)
104 } else {
105 None
106 }
107 }
108}
109
110#[derive(Debug, Clone, PartialEq)]
111pub struct HistoricalDepth {
112 price_levels: BTreeMap<Price, Vec<OrderRun>>,
113 pub aggr_time: Timeframe,
114 tick_size: PriceStep,
115 min_order_qty: MinQtySize,
116 last_snapshot_time: Option<UnixMs>,
117}
118
119pub enum DepthSnapshotUpdate {
120 Stale,
121 Continuous,
122 Gap { start: UnixMs },
123}
124
125impl HistoricalDepth {
126 pub fn new(min_order_qty: MinQtySize, tick_size: PriceStep, basis: Basis) -> Self {
127 Self {
128 price_levels: BTreeMap::new(),
129 aggr_time: match basis {
130 Basis::Time(interval) => interval,
131 Basis::Tick(_) => unimplemented!(),
132 },
133 tick_size,
134 min_order_qty,
135 last_snapshot_time: None,
136 }
137 }
138
139 #[inline]
140 pub fn aggr_time_ms(&self) -> u64 {
141 self.aggr_time.to_milliseconds()
142 }
143
144 pub fn latest_time(&self) -> Option<UnixMs> {
145 self.last_snapshot_time
146 .map(|time| time.floor_to(self.aggr_time))
147 }
148
149 pub fn insert_latest_depth(&mut self, depth: &Depth, time: UnixMs) -> DepthSnapshotUpdate {
150 if let Some(prev_time) = self.last_snapshot_time
151 && time < prev_time
152 {
153 return DepthSnapshotUpdate::Stale;
154 }
155
156 let bucket = time.floor_to(self.aggr_time);
157 let gap_start = self
158 .latest_time()
159 .map(|previous| previous.saturating_add(self.aggr_time_ms()))
160 .filter(|start| *start < bucket);
161
162 self.process_side(&depth.bids, bucket, true, gap_start.is_some());
163 self.process_side(&depth.asks, bucket, false, gap_start.is_some());
164 self.last_snapshot_time = Some(time);
165
166 gap_start.map_or(DepthSnapshotUpdate::Continuous, |start| {
167 DepthSnapshotUpdate::Gap { start }
168 })
169 }
170
171 fn process_side(
172 &mut self,
173 side: &BTreeMap<Price, Qty>,
174 time: UnixMs,
175 is_bid: bool,
176 has_snapshot_gap: bool,
177 ) {
178 let mut current_price = None;
179 let mut current_qty = Qty::ZERO;
180
181 let step = self.tick_size;
182
183 for (price, qty) in side {
184 let rounded_price = price.round_to_side_step(is_bid, step);
185 if Some(rounded_price) == current_price {
186 current_qty += *qty;
187 } else {
188 if let Some(price) = current_price {
189 self.update_price_level(time, price, current_qty, is_bid, has_snapshot_gap);
190 }
191 current_price = Some(rounded_price);
192 current_qty = *qty;
193 }
194 }
195
196 if let Some(price) = current_price {
197 self.update_price_level(time, price, current_qty, is_bid, has_snapshot_gap);
198 }
199 }
200
201 fn update_price_level(
202 &mut self,
203 time: UnixMs,
204 price: Price,
205 qty: Qty,
206 is_bid: bool,
207 has_snapshot_gap: bool,
208 ) {
209 let aggr_time = self.aggr_time;
210 let price_level = self.price_levels.entry(price).or_default();
211
212 match price_level.last_mut() {
213 Some(last_run) if last_run.is_bid == is_bid => {
214 if time > last_run.until_time {
215 if has_snapshot_gap {
216 if qty == last_run.qty {
217 last_run.until_time = time.saturating_add(aggr_time.to_milliseconds());
218 return;
219 }
220
221 last_run.until_time = time;
222 }
223
224 price_level.push(OrderRun::new(time, aggr_time, qty, is_bid));
225 return;
226 }
227
228 if qty == last_run.qty {
229 let new_until = time.saturating_add(aggr_time.to_milliseconds());
230 if new_until > last_run.until_time {
231 last_run.until_time = new_until;
232 }
233 } else {
234 if last_run.until_time > time {
235 last_run.until_time = time;
236 }
237 price_level.push(OrderRun::new(time, aggr_time, qty, is_bid));
238 }
239 }
240 Some(last_run) => {
241 if last_run.until_time > time {
242 last_run.until_time = time;
243 }
244 price_level.push(OrderRun::new(time, aggr_time, qty, is_bid));
245 }
246 None => {
247 price_level.push(OrderRun::new(time, aggr_time, qty, is_bid));
248 }
249 }
250 }
251
252 pub fn is_empty(&self) -> bool {
253 self.price_levels.is_empty()
254 }
255
256 pub fn iter_time_filtered(
257 &self,
258 earliest: UnixMs,
259 latest: UnixMs,
260 highest: Price,
261 lowest: Price,
262 ) -> impl Iterator<Item = (&Price, &Vec<OrderRun>)> {
263 self.price_levels
264 .range(lowest..=highest)
265 .filter(move |(_, runs)| {
266 runs.iter()
267 .any(|run| run.until_time >= earliest && run.start_time <= latest)
268 })
269 }
270
271 pub fn latest_order_runs(
272 &self,
273 highest: Price,
274 lowest: Price,
275 latest_timestamp: UnixMs,
276 ) -> impl Iterator<Item = (&Price, &OrderRun)> {
277 self.price_levels
278 .range(lowest..=highest)
279 .filter_map(move |(price, runs)| {
280 runs.last()
281 .filter(|run| run.until_time >= latest_timestamp)
282 .map(|run| (price, run))
283 })
284 }
285
286 pub fn depth_at(&self, time: UnixMs) -> Depth {
287 let mut depth = Depth::default();
288 for (&price, runs) in &self.price_levels {
289 let first = runs.partition_point(|run| run.until_time <= time);
290 if let Some(run) = runs.get(first)
291 && run.start_time <= time
292 && run.qty > Qty::ZERO
293 {
294 let side = if run.is_bid {
295 &mut depth.bids
296 } else {
297 &mut depth.asks
298 };
299 side.insert(price, run.qty);
300 }
301 }
302 depth
303 }
304
305 pub fn replay_depth_interruptible(
306 &self,
307 start: UnixMs,
308 mut on_snapshot: impl FnMut(UnixMs, &Depth),
309 mut control: impl FnMut() -> ControlFlow<()>,
310 ) -> ControlFlow<()> {
311 #[derive(PartialEq, Eq, PartialOrd, Ord)]
312 enum Change {
313 Remove,
314 Insert,
315 }
316
317 let Some(latest) = self.latest_time().filter(|latest| *latest >= start) else {
318 return ControlFlow::Continue(());
319 };
320
321 let mut cursors = Vec::new();
323 for (&price, runs) in &self.price_levels {
324 control()?;
325 let first = runs.partition_point(|run| run.until_time <= start);
326 let mut remaining = runs[first..]
327 .iter()
328 .take_while(|run| run.start_time <= latest)
329 .filter(|run| run.qty > Qty::ZERO && run.start_time < run.until_time);
330 if let Some(current) = remaining.next() {
331 cursors.push((price, current, remaining));
332 }
333 }
334
335 let mut changes = BTreeMap::<(UnixMs, Change), Vec<usize>>::new();
336 for (index, (_, run, _)) in cursors.iter().enumerate() {
337 control()?;
338 changes
339 .entry((run.start_time.max(start), Change::Insert))
340 .or_default()
341 .push(index);
342 }
343
344 let mut depth = Depth::default();
345 let mut time = start;
346 while let Some(((next, change), indices)) = changes.pop_first() {
347 control()?;
348 if next != time {
349 on_snapshot(time, &depth);
350 time = next;
351 }
352
353 for index in indices {
354 let (price, current, remaining) = &mut cursors[index];
355 let side = if current.is_bid {
356 &mut depth.bids
357 } else {
358 &mut depth.asks
359 };
360 match change {
361 Change::Insert => {
362 side.insert(*price, current.qty);
363 if current.until_time <= latest {
364 changes
365 .entry((current.until_time, Change::Remove))
366 .or_default()
367 .push(index);
368 }
369 }
370 Change::Remove => match remaining.next() {
371 Some(next) if next.start_time == time && next.is_bid == current.is_bid => {
372 side.insert(*price, next.qty);
373 *current = next;
374 if next.until_time <= latest {
375 changes
376 .entry((next.until_time, Change::Remove))
377 .or_default()
378 .push(index);
379 }
380 }
381 Some(next) => {
382 side.remove(price);
383 *current = next;
384 changes
385 .entry((next.start_time, Change::Insert))
386 .or_default()
387 .push(index);
388 }
389 None => {
390 side.remove(price);
391 }
392 },
393 }
394 }
395 }
396
397 control()?;
398 on_snapshot(time, &depth);
399 if time < latest {
400 control()?;
401 on_snapshot(latest, &depth);
402 }
403
404 ControlFlow::Continue(())
405 }
406
407 pub fn cleanup_old_price_levels(&mut self, oldest_time: UnixMs) {
408 self.price_levels.iter_mut().for_each(|(_, runs)| {
409 runs.retain(|run| run.until_time >= oldest_time);
410 });
411
412 self.price_levels.retain(|_, runs| !runs.is_empty());
413 }
414
415 pub fn coalesced_runs(
416 &self,
417 earliest: UnixMs,
418 latest: UnixMs,
419 highest: Price,
420 lowest: Price,
421 market_type: MarketKind,
422 order_size_filter: f32,
423 coalesce_kind: CoalesceKind,
424 ) -> Vec<(Price, OrderRun)> {
425 let mut result_runs = Vec::new();
426
427 let size_in_quote_ccy = volume_size_unit() == SizeUnit::Quote;
428
429 for (price_at_level, runs_at_price_level) in
430 self.iter_time_filtered(earliest, latest, highest, lowest)
431 {
432 let candidate_runs = runs_at_price_level
433 .iter()
434 .filter(|run_ref| {
435 if !(run_ref.until_time >= earliest && run_ref.start_time <= latest) {
436 return false;
437 }
438 let order_size = market_type.qty_in_quote_value(
439 run_ref.qty,
440 *price_at_level,
441 size_in_quote_ccy,
442 );
443 order_size as f32 > order_size_filter
444 })
445 .collect::<Vec<&OrderRun>>();
446
447 if candidate_runs.is_empty() {
448 continue;
449 }
450
451 let mut current_accumulator_opt: Option<CoalescingRun> = None;
452
453 for run_to_process_ref in candidate_runs {
454 let run_to_process = *run_to_process_ref;
455
456 if let Some(current_accumulator) = current_accumulator_opt.as_mut() {
457 let comparison_base_qty = current_accumulator.comparison_qty(&coalesce_kind);
458 let qty_within_threshold = coalesce_kind.is_within_lot_similarity(
459 comparison_base_qty,
460 run_to_process.qty,
461 self.min_order_qty,
462 );
463
464 if run_to_process.start_time <= current_accumulator.until_time
465 && run_to_process.is_bid == current_accumulator.is_bid
466 && qty_within_threshold
467 {
468 current_accumulator.merge_run(&run_to_process);
469 } else {
470 result_runs.push((
471 *price_at_level,
472 current_accumulator.to_order_run(&coalesce_kind),
473 ));
474 current_accumulator_opt = Some(CoalescingRun::new(&run_to_process));
475 }
476 } else {
477 current_accumulator_opt = Some(CoalescingRun::new(&run_to_process));
478 }
479 }
480
481 if let Some(accumulator) = current_accumulator_opt {
482 result_runs.push((*price_at_level, accumulator.to_order_run(&coalesce_kind)));
483 }
484 }
485 result_runs
486 }
487
488 pub fn query_grid_qtys(
489 &self,
490 center_time: UnixMs,
491 center_price: Price,
492 time_interval_offsets: &[i64],
493 price_tick_offsets: &[i64],
494 market_type: MarketKind,
495 order_size_filter: f32,
496 coalesce_kind: Option<CoalesceKind>,
497 ) -> FxHashMap<(UnixMs, Price), (Qty, bool)> {
498 let aggr_time_ms = self.aggr_time_ms();
499 let aggr_time = self.aggr_time;
500
501 let step = self.tick_size;
502 let target_prices: Vec<Price> = price_tick_offsets
503 .iter()
504 .copied()
505 .filter_map(|offset| center_price.checked_add_steps(offset, step))
506 .collect();
507 let capacity = time_interval_offsets.len() * target_prices.len();
508 if target_prices.is_empty() {
509 return FxHashMap::with_capacity_and_hasher(capacity, FxBuildHasher);
510 }
511
512 let query_earliest_time = time_interval_offsets
513 .iter()
514 .map(|offset| center_time.offset_by_timeframe(aggr_time, *offset))
515 .min()
516 .unwrap_or(center_time);
517
518 let query_latest_time = time_interval_offsets
519 .iter()
520 .map(|offset| center_time.offset_by_timeframe(aggr_time, *offset))
521 .max()
522 .map_or(center_time, |t| t.saturating_add(aggr_time_ms));
523
524 let query_lowest = target_prices.iter().copied().min().unwrap_or(center_price);
525 let query_highest = target_prices.iter().copied().max().unwrap_or(center_price);
526
527 let runs_in_vicinity: Vec<(Price, OrderRun)> = if let Some(ck) = coalesce_kind {
528 self.coalesced_runs(
529 query_earliest_time,
530 query_latest_time,
531 query_highest,
532 query_lowest,
533 market_type,
534 order_size_filter,
535 ck,
536 )
537 } else {
538 self.iter_time_filtered(
539 query_earliest_time,
540 query_latest_time,
541 query_highest,
542 query_lowest,
543 )
544 .flat_map(|(price_level, runs_at_price)| {
545 runs_at_price.iter().map(move |run| (*price_level, *run))
546 })
547 .collect()
548 };
549
550 let mut grid_quantities: FxHashMap<(UnixMs, Price), (Qty, bool)> =
551 FxHashMap::with_capacity_and_hasher(capacity, FxBuildHasher);
552 for target_price_key in target_prices {
553 for time_offset in time_interval_offsets {
554 let target_time_val = center_time.offset_by_timeframe(aggr_time, *time_offset);
555 let current_grid_key = (target_time_val, target_price_key);
556
557 for (run_price_level, run_data) in &runs_in_vicinity {
558 if *run_price_level == target_price_key
559 && run_data.start_time <= target_time_val
560 && run_data.until_time > target_time_val
561 {
562 grid_quantities.insert(current_grid_key, (run_data.qty, run_data.is_bid));
563 break;
564 }
565 }
566 }
567 }
568 grid_quantities
569 }
570
571 pub fn max_qty_in_range_raw(
572 &self,
573 earliest: UnixMs,
574 latest: UnixMs,
575 highest: Price,
576 lowest: Price,
577 ) -> Qty {
578 let mut max_qty = Qty::ZERO;
579
580 for (_price, runs) in self.price_levels.range(lowest..=highest) {
581 for run in runs.iter() {
582 if run.until_time < earliest || run.start_time > latest {
583 continue;
584 }
585 max_qty = max_qty.max(run.qty);
586 }
587 }
588
589 max_qty
590 }
591
592 pub fn max_depth_qty_in_range(
593 &self,
594 earliest: UnixMs,
595 latest: UnixMs,
596 highest: Price,
597 lowest: Price,
598 market_type: MarketKind,
599 order_size_filter: f32,
600 ) -> Qty {
601 let mut max_depth_qty = Qty::ZERO;
602 let size_in_quote_ccy = volume_size_unit() == SizeUnit::Quote;
603
604 for (price, runs) in self.price_levels.range(lowest..=highest) {
605 for run in runs.iter() {
606 if run.until_time < earliest || run.start_time > latest {
607 continue;
608 }
609
610 let order_size = market_type.qty_in_quote_value(run.qty, *price, size_in_quote_ccy);
611
612 if order_size as f32 > order_size_filter {
613 max_depth_qty = max_depth_qty.max(run.qty);
614 }
615 }
616 }
617
618 max_depth_qty
619 }
620}
621
622#[derive(Debug, Clone, Copy, Deserialize, Serialize)]
623pub enum CoalesceKind {
624 First(f32),
625 Average(f32),
626 Max(f32),
627}
628
629impl CoalesceKind {
630 pub fn threshold(&self) -> f32 {
631 match self {
632 CoalesceKind::Average(t) | CoalesceKind::First(t) | CoalesceKind::Max(t) => *t,
633 }
634 }
635
636 pub fn with_threshold(&self, threshold: f32) -> Self {
637 match self {
638 CoalesceKind::First(_) => CoalesceKind::First(threshold),
639 CoalesceKind::Average(_) => CoalesceKind::Average(threshold),
640 CoalesceKind::Max(_) => CoalesceKind::Max(threshold),
641 }
642 }
643
644 fn is_within_lot_similarity(
645 self,
646 base_qty: Qty,
647 candidate_qty: Qty,
648 min_qty: MinQtySize,
649 ) -> bool {
650 let ratio = self.threshold().max(0.0);
651
652 if !ratio.is_finite() {
653 return false;
654 }
655
656 let base_lots = base_qty.to_lots(min_qty).max(0);
657 let candidate_lots = candidate_qty.to_lots(min_qty).max(0);
658
659 if base_lots == 0 {
660 return candidate_lots == 0;
661 }
662
663 let lots_diff = base_lots.abs_diff(candidate_lots) as f64;
664 let allowed_diff = (base_lots as f64) * (ratio as f64);
665
666 lots_diff <= allowed_diff
667 }
668}
669
670impl PartialEq for CoalesceKind {
671 fn eq(&self, other: &Self) -> bool {
672 std::mem::discriminant(self) == std::mem::discriminant(other)
673 }
674}
675
676impl Eq for CoalesceKind {}
677
678#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq)]
679pub struct CoalescingRun {
680 pub start_time: UnixMs,
681 pub until_time: UnixMs,
682 pub is_bid: bool,
683 pub qty_sum: Qty,
684 pub run_count: u32,
685 first_qty: Qty,
686 max_qty: Qty,
687}
688
689impl CoalescingRun {
690 pub fn new(run: &OrderRun) -> Self {
691 let run_qty = run.qty;
692 CoalescingRun {
693 start_time: run.start_time,
694 until_time: run.until_time,
695 is_bid: run.is_bid,
696 qty_sum: run_qty,
697 run_count: 1,
698 first_qty: run_qty,
699 max_qty: run_qty,
700 }
701 }
702
703 pub fn merge_run(&mut self, run: &OrderRun) {
704 self.until_time = self.until_time.max(run.until_time);
705 let run_qty = run.qty;
706 self.qty_sum += run_qty;
707 self.run_count += 1;
708 self.max_qty = self.max_qty.max(run_qty);
709 }
710
711 pub fn comparison_qty(&self, kind: &CoalesceKind) -> Qty {
712 match kind {
713 CoalesceKind::Average(_) => self.current_average_qty(),
714 CoalesceKind::Max(_) | CoalesceKind::First(_) => self.first_qty,
715 }
716 }
717
718 pub fn current_average_qty(&self) -> Qty {
719 if self.run_count == 0 {
720 Qty::ZERO
721 } else {
722 let count = i64::from(self.run_count);
723 let rounded_units = (self.qty_sum.units + (count / 2)) / count;
724 Qty::from_units(rounded_units)
725 }
726 }
727
728 pub fn to_order_run(&self, kind: &CoalesceKind) -> OrderRun {
729 let final_qty = match kind {
730 CoalesceKind::Average(_) => self.current_average_qty(),
731 CoalesceKind::First(_) => self.first_qty,
732 CoalesceKind::Max(_) => self.max_qty,
733 };
734 OrderRun {
735 start_time: self.start_time,
736 until_time: self.until_time,
737 qty: final_qty,
738 is_bid: self.is_bid,
739 }
740 }
741}
742
743#[derive(Debug, Clone)]
744pub struct GroupedTrade {
745 pub is_sell: bool,
746 pub price: Price,
747 pub qty: Qty,
748}
749
750impl GroupedTrade {
751 pub fn compare_with(&self, price: Price, is_sell: bool) -> std::cmp::Ordering {
752 if self.is_sell == is_sell {
753 self.price.cmp(&price)
754 } else {
755 self.is_sell.cmp(&is_sell)
756 }
757 }
758}
759
760#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
761pub enum ProfileKind {
762 FixedWindow(usize),
763 #[default]
764 VisibleRange,
765}
766
767impl std::fmt::Display for ProfileKind {
768 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
769 match self {
770 ProfileKind::FixedWindow(_) => write!(f, "Fixed window"),
771 ProfileKind::VisibleRange => write!(f, "Visible range"),
772 }
773 }
774}