1use std::collections::BTreeMap;
2
3use crate::chart::Basis;
4use crate::chart::heatmap::HeatmapDataPoint;
5use crate::chart::kline::{ClusterKind, KlineDataPoint, KlineTrades, NPoc};
6
7use exchange::unit::{Price, PriceStep, Qty};
8use exchange::{Kline, Timeframe, Trade, Volume};
9
10pub trait DataPoint {
11 fn add_trade(&mut self, trade: &Trade, step: PriceStep);
12
13 fn clear_trades(&mut self);
14
15 fn last_trade_time(&self) -> Option<u64>;
16
17 fn first_trade_time(&self) -> Option<u64>;
18
19 fn last_price(&self) -> Price;
20
21 fn kline(&self) -> Option<&Kline>;
22
23 fn value_high(&self) -> Price;
24
25 fn value_low(&self) -> Price;
26}
27
28pub struct TimeSeries<D: DataPoint> {
29 pub datapoints: BTreeMap<u64, D>,
30 pub interval: Timeframe,
31 pub tick_size: PriceStep,
32}
33
34impl<D: DataPoint> TimeSeries<D> {
35 pub fn base_price(&self) -> Price {
36 self.datapoints
37 .values()
38 .last()
39 .map_or(Price::from_f32(0.0), DataPoint::last_price)
40 }
41
42 pub fn latest_timestamp(&self) -> Option<u64> {
43 self.datapoints.keys().last().copied()
44 }
45
46 pub fn latest_kline(&self) -> Option<&Kline> {
47 self.datapoints.values().last().and_then(|dp| dp.kline())
48 }
49
50 pub fn price_scale(&self, lookback: usize) -> (Price, Price) {
51 let mut iter = self.datapoints.iter().rev().take(lookback);
52
53 if let Some((_, first)) = iter.next() {
54 let mut high = first.value_high();
55 let mut low = first.value_low();
56
57 for (_, dp) in iter {
58 let value_high = dp.value_high();
59 let value_low = dp.value_low();
60 if value_high > high {
61 high = value_high;
62 }
63 if value_low < low {
64 low = value_low;
65 }
66 }
67
68 (high, low)
69 } else {
70 (Price::from_f32(0.0), Price::from_f32(0.0))
71 }
72 }
73
74 pub fn volume_data<'a>(&'a self) -> BTreeMap<u64, exchange::Volume>
75 where
76 BTreeMap<u64, exchange::Volume>: From<&'a TimeSeries<D>>,
77 {
78 self.into()
79 }
80
81 pub fn timerange(&self) -> (u64, u64) {
82 let earliest = self.datapoints.keys().next().copied().unwrap_or(0);
83 let latest = self.datapoints.keys().last().copied().unwrap_or(0);
84
85 (earliest, latest)
86 }
87
88 pub fn min_max_price_in_range_prices(
89 &self,
90 earliest: u64,
91 latest: u64,
92 ) -> Option<(Price, Price)> {
93 let mut it = self.datapoints.range(earliest..=latest);
94
95 let (_, first) = it.next()?;
96 let mut min_price = first.value_low();
97 let mut max_price = first.value_high();
98
99 for (_, dp) in it {
100 let low = dp.value_low();
101 let high = dp.value_high();
102 if low < min_price {
103 min_price = low;
104 }
105 if high > max_price {
106 max_price = high;
107 }
108 }
109
110 Some((min_price, max_price))
111 }
112
113 pub fn min_max_price_in_range(&self, earliest: u64, latest: u64) -> Option<(f32, f32)> {
114 self.min_max_price_in_range_prices(earliest, latest)
115 .map(|(min_p, max_p)| (min_p.to_f32(), max_p.to_f32()))
116 }
117
118 pub fn clear_trades(&mut self) {
119 for data_point in self.datapoints.values_mut() {
120 data_point.clear_trades();
121 }
122 }
123
124 fn align_down_to_phase(time: u64, phase: u64, interval: u64) -> u64 {
125 if time >= phase {
126 time.saturating_sub((time - phase) % interval)
127 } else {
128 phase
129 }
130 }
131
132 fn check_kline_integrity_range(
133 &self,
134 earliest: u64,
135 latest: u64,
136 interval: u64,
137 ) -> Option<Vec<u64>> {
138 let mut time = earliest;
139 let mut missing_count = 0;
140
141 while time < latest {
142 if !self.datapoints.contains_key(&time) {
143 missing_count += 1;
144 break;
145 }
146 time += interval;
147 }
148
149 if missing_count > 0 {
150 let mut missing_keys = Vec::with_capacity(((latest - earliest) / interval) as usize);
151 let mut time = earliest;
152
153 while time < latest {
154 if !self.datapoints.contains_key(&time) {
155 missing_keys.push(time);
156 }
157 time += interval;
158 }
159
160 log::debug!(
161 "Integrity check failed: missing {} klines",
162 missing_keys.len()
163 );
164 return Some(missing_keys);
165 }
166
167 None
168 }
169
170 pub fn check_kline_integrity(&self, earliest: u64, latest: u64) -> Option<Vec<u64>> {
171 if self.datapoints.is_empty() {
172 return None;
173 }
174
175 let interval = self.interval.to_milliseconds();
176 if interval == 0 {
177 return None;
178 }
179
180 let (series_earliest, series_latest) = self.timerange();
181 let phase = series_earliest % interval;
182
183 let check_earliest =
184 Self::align_down_to_phase(earliest.max(series_earliest), phase, interval)
185 .max(series_earliest);
186 let check_latest = Self::align_down_to_phase(latest.min(series_latest), phase, interval)
187 .min(series_latest);
188
189 if check_earliest < check_latest {
190 self.check_kline_integrity_range(check_earliest, check_latest, interval)
191 } else {
192 None
193 }
194 }
195}
196
197impl TimeSeries<KlineDataPoint> {
198 pub fn new(interval: Timeframe, tick_size: PriceStep, klines: &[Kline]) -> Self {
199 let mut timeseries = Self {
200 datapoints: BTreeMap::new(),
201 interval,
202 tick_size,
203 };
204
205 timeseries.insert_klines(klines);
206 timeseries
207 }
208
209 pub fn with_trades(&self, trades: &[Trade]) -> TimeSeries<KlineDataPoint> {
210 let mut new_series = Self {
211 datapoints: self.datapoints.clone(),
212 interval: self.interval,
213 tick_size: self.tick_size,
214 };
215
216 new_series.insert_trades_or_create_bucket(trades);
217 new_series
218 }
219
220 pub fn insert_klines(&mut self, klines: &[Kline]) {
221 for kline in klines {
222 let entry = self
223 .datapoints
224 .entry(kline.time)
225 .or_insert_with(|| KlineDataPoint {
226 kline: *kline,
227 footprint: KlineTrades::new(),
228 });
229
230 entry.kline = *kline;
231 }
232
233 self.update_poc_status();
234 }
235
236 pub fn insert_trades_or_create_bucket(&mut self, buffer: &[Trade]) {
237 if buffer.is_empty() {
238 return;
239 }
240 let aggr_time = self.interval.to_milliseconds();
241 let mut updated_times = Vec::new();
242
243 buffer.iter().for_each(|trade| {
244 let rounded_time = (trade.time / aggr_time) * aggr_time;
245
246 if !updated_times.contains(&rounded_time) {
247 updated_times.push(rounded_time);
248 }
249
250 let entry = self
251 .datapoints
252 .entry(rounded_time)
253 .or_insert_with(|| KlineDataPoint {
254 kline: Kline {
255 time: rounded_time,
256 open: trade.price,
257 high: trade.price,
258 low: trade.price,
259 close: trade.price,
260 volume: Volume::empty_buy_sell(),
261 },
262 footprint: KlineTrades::new(),
263 });
264
265 entry.add_trade(trade, self.tick_size);
266 });
267
268 for time in updated_times {
269 if let Some(data_point) = self.datapoints.get_mut(&time) {
270 data_point.calculate_poc();
271 }
272 }
273 }
274
275 pub fn insert_trades_existing_buckets(&mut self, buffer: &[Trade]) {
276 if buffer.is_empty() {
277 return;
278 }
279 let aggr_time = self.interval.to_milliseconds();
280 let mut updated_times: Vec<u64> = Vec::new();
281
282 for trade in buffer {
283 let rounded_time = (trade.time / aggr_time) * aggr_time;
284
285 if let Some(entry) = self.datapoints.get_mut(&rounded_time) {
286 if !updated_times.contains(&rounded_time) {
287 updated_times.push(rounded_time);
288 }
289 entry.add_trade(trade, self.tick_size);
290 }
291 }
292
293 for time in updated_times {
294 if let Some(data_point) = self.datapoints.get_mut(&time) {
295 data_point.calculate_poc();
296 }
297 }
298 }
299
300 pub fn change_tick_size(&mut self, tick_size: PriceStep, raw_trades: &[Trade]) {
301 self.tick_size = tick_size;
302
303 self.clear_trades();
304
305 if !raw_trades.is_empty() {
306 self.insert_trades_existing_buckets(raw_trades);
307 }
308 }
309
310 pub fn update_poc_status(&mut self) {
311 let updates = self
312 .datapoints
313 .iter()
314 .filter_map(|(&time, dp)| dp.poc_price().map(|price| (time, price)))
315 .collect::<Vec<_>>();
316
317 for (current_time, poc_price) in updates {
318 let mut npoc = NPoc::default();
319
320 for (&next_time, next_dp) in self.datapoints.range((current_time + 1)..) {
321 let next_dp_low = next_dp.kline.low.round_to_side_step(true, self.tick_size);
322 let next_dp_high = next_dp.kline.high.round_to_side_step(false, self.tick_size);
323
324 if next_dp_low <= poc_price && next_dp_high >= poc_price {
325 npoc.filled(next_time);
326 break;
327 } else {
328 npoc.unfilled();
329 }
330 }
331
332 if let Some(data_point) = self.datapoints.get_mut(¤t_time) {
333 data_point.set_poc_status(npoc);
334 }
335 }
336 }
337
338 pub fn suggest_trade_fetch_range(
339 &self,
340 visible_earliest: u64,
341 visible_latest: u64,
342 ) -> Option<(u64, u64)> {
343 if self.datapoints.is_empty() {
344 return None;
345 }
346
347 self.find_trade_gap()
348 .and_then(|(last_t_before_gap, first_t_after_gap)| {
349 if last_t_before_gap.is_none() && first_t_after_gap.is_none() {
350 return None;
351 }
352 let (data_earliest, data_latest) = self.timerange();
353
354 let fetch_from = last_t_before_gap
355 .map_or(data_earliest, |t| t.saturating_add(1))
356 .max(visible_earliest);
357 let fetch_to = first_t_after_gap
358 .map_or(data_latest, |t| t.saturating_sub(1))
359 .min(visible_latest);
360
361 if fetch_from < fetch_to {
362 Some((fetch_from, fetch_to))
363 } else {
364 None
365 }
366 })
367 }
368
369 fn find_trade_gap(&self) -> Option<(Option<u64>, Option<u64>)> {
370 let empty_kline_time = self
371 .datapoints
372 .iter()
373 .rev()
374 .find(|(_, dp)| dp.footprint.trades.is_empty())
375 .map(|(&time, _)| time);
376
377 if let Some(target_time) = empty_kline_time {
378 let last_t_before_gap = self
379 .datapoints
380 .range(..target_time)
381 .rev()
382 .find_map(|(_, dp)| dp.last_trade_time());
383
384 let first_t_after_gap = self
385 .datapoints
386 .range(target_time + 1..)
387 .find_map(|(_, dp)| dp.first_trade_time());
388
389 Some((last_t_before_gap, first_t_after_gap))
390 } else {
391 None
392 }
393 }
394
395 pub fn max_qty_ts_range(
396 &self,
397 cluster_kind: ClusterKind,
398 earliest: u64,
399 latest: u64,
400 highest: Price,
401 lowest: Price,
402 ) -> Qty {
403 let mut max_cluster_qty: Qty = Qty::default();
404
405 self.datapoints
406 .range(earliest..=latest)
407 .for_each(|(_, dp)| {
408 max_cluster_qty =
409 max_cluster_qty.max(dp.max_cluster_qty(cluster_kind, highest, lowest));
410 });
411
412 max_cluster_qty
413 }
414}
415
416impl TimeSeries<HeatmapDataPoint> {
417 pub fn new(basis: Basis, tick_size: PriceStep) -> Self {
418 let timeframe = match basis {
419 Basis::Time(interval) => interval,
420 Basis::Tick(_) => unimplemented!(),
421 };
422
423 Self {
424 datapoints: BTreeMap::new(),
425 interval: timeframe,
426 tick_size,
427 }
428 }
429
430 pub fn max_trade_qty_and_aggr_volume(&self, earliest: u64, latest: u64) -> (Qty, Qty) {
431 let mut max_trade_qty = Qty::ZERO;
432 let mut max_aggr_volume = Qty::ZERO;
433
434 self.datapoints
435 .range(earliest..=latest)
436 .for_each(|(_, dp)| {
437 let (mut buy_volume, mut sell_volume) = (Qty::ZERO, Qty::ZERO);
438
439 dp.grouped_trades.iter().for_each(|trade| {
440 let trade_qty = trade.qty;
441 max_trade_qty = max_trade_qty.max(trade_qty);
442
443 if trade.is_sell {
444 sell_volume += trade_qty;
445 } else {
446 buy_volume += trade_qty;
447 }
448 });
449
450 max_aggr_volume = max_aggr_volume.max(buy_volume + sell_volume);
451 });
452
453 (max_trade_qty, max_aggr_volume)
454 }
455}
456
457impl From<&TimeSeries<KlineDataPoint>> for BTreeMap<u64, exchange::Volume> {
458 fn from(timeseries: &TimeSeries<KlineDataPoint>) -> Self {
460 timeseries
461 .datapoints
462 .iter()
463 .map(|(time, dp)| (*time, dp.kline.volume))
464 .collect()
465 }
466}