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