1use super::indicators::{
2 IndicatorHost, IndicatorInputCoverage, IndicatorInputSource, OpenInterestSeries,
3};
4use super::series::KlineSeries;
5use super::{
6 IndicatorInstanceId, IndicatorInstanceStore, IndicatorUpdateScope, KlineIndexChange,
7 KlineStudy, KlineUpdate, LineStrokeStyle, MIN_LINE_WIDTH, SeriesColor, StudyDataSource,
8 ThemeAccent,
9};
10use crate::chart::Basis;
11use exchange::{Kline, TickMultiplier, TickerInfo, Timeframe, Trade, UnixMs, unit::PriceStep};
12use indicator_engine::IndicatorFrame;
13use rustc_hash::FxHashMap;
14use std::sync::Arc;
15
16pub struct LiveIndicatorUpdate {
17 pub committed_bar: bool,
18 pub rebuild_scope: Option<IndicatorUpdateScope>,
19 pub closed_trade_correction: bool,
20 pub coverage_frames: Vec<IndicatorFrame>,
21 pub index_change: KlineIndexChange,
22}
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum HistoricalInputUpdate {
26 Rejected,
27 NoData { has_consumer: bool },
28 Ignored,
29 Applied,
30}
31
32impl HistoricalInputUpdate {
33 pub fn needs_rebuild(self) -> bool {
34 matches!(self, Self::NoData { has_consumer: true } | Self::Applied)
35 }
36}
37
38pub struct KlineData {
39 series: Vec<KlineSeries>,
40 indicator_host: IndicatorHost,
41 base_ticker: TickerInfo,
42 series_index: FxHashMap<TickerInfo, usize>,
43 tick_multiplier: TickMultiplier,
44}
45
46impl KlineData {
47 pub fn new(base_ticker: TickerInfo, tick_multiplier: TickMultiplier) -> Self {
48 Self {
49 series: Vec::new(),
50 indicator_host: IndicatorHost::default(),
51 base_ticker,
52 series_index: FxHashMap::default(),
53 tick_multiplier,
54 }
55 }
56
57 pub fn base_ticker(&self) -> TickerInfo {
58 self.base_ticker
59 }
60
61 pub fn selected_tickers(&self) -> Vec<TickerInfo> {
62 self.series
63 .iter()
64 .map(|series| series.ticker_info)
65 .collect()
66 }
67
68 pub fn series(&self) -> &[KlineSeries] {
69 &self.series
70 }
71
72 pub fn indicator_host(&self) -> &IndicatorHost {
73 &self.indicator_host
74 }
75
76 pub fn merge_indicator_host(
77 &mut self,
78 rebuilt: IndicatorHost,
79 replaced_ids: &[IndicatorInstanceId],
80 active_ids: &[IndicatorInstanceId],
81 indicators: &IndicatorInstanceStore,
82 ) {
83 self.indicator_host
84 .merge_rebuilt(rebuilt, replaced_ids, active_ids, indicators);
85 }
86
87 pub fn remap_indicator_outputs(&mut self, change: KlineIndexChange) {
88 match change {
89 KlineIndexChange::None => {}
90 KlineIndexChange::Prefix(prefix) => self.indicator_host.remap_outputs_prefix(prefix),
91 KlineIndexChange::Explicit(previous_indices) => {
92 self.indicator_host.remap_outputs(&previous_indices)
93 }
94 }
95 }
96
97 pub fn dependent_indicator_ids(
98 &self,
99 roots: &[IndicatorInstanceId],
100 ) -> Vec<IndicatorInstanceId> {
101 self.indicator_host.dependent_ids(roots)
102 }
103
104 pub fn calculation_indicator_ids(
105 &self,
106 roots: &[IndicatorInstanceId],
107 ) -> Vec<IndicatorInstanceId> {
108 self.indicator_host.calculation_ids(roots)
109 }
110
111 pub fn has_indicator_entries(&self, ids: &[IndicatorInstanceId]) -> bool {
112 self.indicator_host.has_entries(ids)
113 }
114
115 pub fn tick_multiplier(&self) -> TickMultiplier {
116 self.tick_multiplier
117 }
118
119 pub fn base_series(&self) -> Option<&KlineSeries> {
120 let idx = self.series_index.get(&self.base_ticker).copied()?;
121 self.series.get(idx)
122 }
123
124 fn base_series_mut(&mut self) -> Option<&mut KlineSeries> {
125 let idx = self.series_index.get(&self.base_ticker).copied()?;
126 self.series.get_mut(idx)
127 }
128
129 pub fn configure_recurring_profile_intervals(&mut self, intervals: &[(Timeframe, PriceStep)]) {
130 if let Some(base_series) = self.base_series_mut() {
131 base_series.configure_recurring_profile_intervals(intervals);
132 }
133 }
134
135 pub fn prepare_trade_price_distribution(&mut self, earliest: UnixMs, latest: UnixMs) {
136 let tick_size = self.tick_size();
137 if let Some(series) = self.base_series_mut() {
138 series.prepare_trade_price_distribution(earliest, latest, tick_size);
139 }
140 }
141
142 pub fn configure_big_trades_threshold(&mut self, threshold: Option<u64>) {
143 if let Some(base_series) = self.base_series_mut() {
144 base_series.configure_big_trades_threshold(threshold);
145 }
146 }
147
148 pub fn set_tick_multiplier(&mut self, multiplier: TickMultiplier) -> bool {
149 if self.tick_multiplier == multiplier {
150 return false;
151 }
152 self.tick_multiplier = multiplier;
153 true
154 }
155
156 pub fn tick_size(&self) -> PriceStep {
157 self.tick_multiplier
158 .multiply_with_min_tick_step(self.base_ticker)
159 }
160
161 pub fn set_series_color(&mut self, ticker_info: TickerInfo, color: SeriesColor) -> bool {
162 let Some(idx) = self.series_index.get(&ticker_info).copied() else {
163 return false;
164 };
165
166 self.series[idx].set_series_color(color)
167 }
168
169 pub fn set_series_line_width(&mut self, ticker_info: TickerInfo, width: u8) -> bool {
170 let Some(idx) = self.series_index.get(&ticker_info).copied() else {
171 return false;
172 };
173
174 self.series[idx].set_line_width(width)
175 }
176
177 pub fn set_series_stroke_style(
178 &mut self,
179 ticker_info: TickerInfo,
180 stroke_style: LineStrokeStyle,
181 ) -> bool {
182 let Some(idx) = self.series_index.get(&ticker_info).copied() else {
183 return false;
184 };
185
186 self.series[idx].set_stroke_style(stroke_style)
187 }
188
189 pub fn reset_series_style(&mut self, ticker_info: TickerInfo) -> bool {
190 let Some(index) = self.series_index.get(&ticker_info).copied() else {
191 return false;
192 };
193
194 let color = self.default_series_color(index);
195 let series = &mut self.series[index];
196 let color_changed = series.set_series_color(color);
197 let width_changed = series.set_line_width(MIN_LINE_WIDTH);
198 let stroke_changed = series.set_stroke_style(LineStrokeStyle::Solid);
199
200 color_changed || width_changed || stroke_changed
201 }
202
203 fn default_series_color(&self, index: usize) -> SeriesColor {
204 if self.series[index].ticker_info == self.base_ticker {
205 return SeriesColor::default();
206 }
207
208 SeriesColor::with_accent(self.next_unused_accent(Some(index)))
209 }
210
211 pub fn add_ticker(&mut self, ticker_info: TickerInfo, basis: Basis) -> bool {
212 if self.series_index.contains_key(&ticker_info) {
213 return false;
214 }
215
216 let idx = self.series.len();
217 let tick_size = self
218 .tick_multiplier
219 .multiply_with_min_tick_step(ticker_info);
220 let color = if ticker_info == self.base_ticker {
221 SeriesColor::default()
222 } else {
223 SeriesColor::with_accent(self.next_unused_accent(None))
224 };
225 let series = KlineSeries::new_with_basis(ticker_info, color, basis, tick_size);
226
227 self.series.push(series);
228 self.series_index.insert(ticker_info, idx);
229 true
230 }
231
232 fn next_unused_accent(&self, excluding: Option<usize>) -> ThemeAccent {
235 let is_used = |accent: ThemeAccent| {
236 self.series.iter().enumerate().any(|(index, series)| {
237 Some(index) != excluding && series.color().accent_value() == Some(accent)
238 })
239 };
240
241 ThemeAccent::ALL
242 .into_iter()
243 .find(|accent| !is_used(*accent))
244 .unwrap_or(ThemeAccent::One)
245 }
246
247 pub fn remove_ticker(&mut self, ticker_info: TickerInfo) -> bool {
248 let Some(idx) = self.series_index.remove(&ticker_info) else {
249 return false;
250 };
251
252 self.series.remove(idx);
253 self.rebuild_series_index();
254 true
255 }
256
257 fn rebuild_series_index(&mut self) {
258 self.series_index.clear();
259 for (idx, series) in self.series.iter().enumerate() {
260 self.series_index.insert(series.ticker_info, idx);
261 }
262 }
263
264 pub fn reset_for_basis(&mut self, basis: Basis, preserve_raw_trades: bool) {
265 for series in &mut self.series {
266 series.set_aggregation_basis(basis, preserve_raw_trades);
267 }
268 }
269
270 pub fn configure_series_for_basis(&mut self, basis: Basis) {
271 for series in &mut self.series {
272 let tick_size = self
273 .tick_multiplier
274 .multiply_with_min_tick_step(series.ticker_info);
275 if series.basis() != basis {
276 series.set_aggregation_basis(basis, true);
277 }
278 series.set_tick_size(tick_size);
279 }
280 }
281
282 pub fn configure_indicators(
283 &mut self,
284 indicators: &IndicatorInstanceStore,
285 active_indicators: &[IndicatorInstanceId],
286 ) -> Vec<IndicatorInstanceId> {
287 self.indicator_host
288 .set_active(active_indicators, indicators)
289 }
290
291 pub fn can_recalculate_incrementally(
292 &self,
293 ids: &[IndicatorInstanceId],
294 scope: IndicatorUpdateScope,
295 ) -> bool {
296 let IndicatorUpdateScope::From(start) = scope else {
297 return false;
298 };
299 self.indicator_host.can_append_from(ids, start)
300 && self.base_series().is_none_or(|series| {
301 series
302 .committed_points()
303 .filter(|point| point.index >= start)
304 .take(2)
305 .count()
306 <= 1
307 })
308 }
309
310 pub fn append_indicator_frames(
311 &mut self,
312 indicator_ids: &[IndicatorInstanceId],
313 start: usize,
314 trade_data_enabled: bool,
315 ) -> Arc<[IndicatorFrame]> {
316 let frames = self.evaluate_closed_append_from(start, indicator_ids, trade_data_enabled);
317 self.preview_indicator_open_for(indicator_ids, trade_data_enabled);
318 frames.into()
319 }
320
321 pub fn committed_indicator_frames(&self, trade_data_enabled: bool) -> Arc<[IndicatorFrame]> {
322 let Some(base_series) = self.base_series() else {
323 return Arc::from([]);
324 };
325 let frames = base_series
326 .committed_points()
327 .map(|point| self.indicator_frame(point.index, point.kline.time, trade_data_enabled))
328 .collect::<Vec<_>>();
329 frames.into()
330 }
331
332 pub fn latest_indicator_frame(&self, trade_data_enabled: bool) -> Option<IndicatorFrame> {
333 let base_series = self.base_series()?;
334 let point = base_series.latest_point()?;
335 Some(self.indicator_frame(point.index, point.kline.time, trade_data_enabled))
336 }
337
338 pub fn indicator_input_coverage_for_frames(
339 &self,
340 committed_frames: &[IndicatorFrame],
341 trade_data_enabled: bool,
342 ) -> IndicatorInputCoverage {
343 let mut frames = committed_frames.to_vec();
344 if let Some(latest) = self.latest_indicator_frame(trade_data_enabled) {
345 if let Some(existing) = frames.iter_mut().find(|frame| frame.index == latest.index) {
346 *existing = latest;
347 } else {
348 frames.push(latest);
349 }
350 }
351 IndicatorInputCoverage::from_frames(&frames)
352 }
353
354 pub fn preview_indicator_open_for(
355 &mut self,
356 ids: &[IndicatorInstanceId],
357 trade_data_enabled: bool,
358 ) -> bool {
359 let Some(frame) = self.latest_indicator_frame(trade_data_enabled) else {
360 return false;
361 };
362 self.indicator_host.preview_open_selected(ids, &frame);
363 true
364 }
365
366 fn evaluate_closed_previous_frame(
367 &mut self,
368 time: Option<exchange::UnixMs>,
369 indicator_ids: &[IndicatorInstanceId],
370 trade_data_enabled: bool,
371 ) -> Option<IndicatorFrame> {
372 let time = time?;
373 let base_series = self.base_series()?;
374
375 let committed_len = base_series.point_count().saturating_sub(1);
376 let point = base_series
377 .points()
378 .take(committed_len)
379 .next_back()
380 .filter(|point| point.kline.time == time)?;
381
382 let frame = self.indicator_frame(point.index, point.kline.time, trade_data_enabled);
383
384 self.indicator_host
385 .evaluate_closed_selected(indicator_ids, &frame);
386 Some(frame)
387 }
388
389 fn evaluate_closed_append_from(
390 &mut self,
391 start: usize,
392 indicator_ids: &[IndicatorInstanceId],
393 trade_data_enabled: bool,
394 ) -> Vec<IndicatorFrame> {
395 let points = self
396 .base_series()
397 .map(|series| {
398 series
399 .committed_points()
400 .skip(start)
401 .map(|point| (point.index, point.kline.time))
402 .collect::<Vec<_>>()
403 })
404 .unwrap_or_default();
405 let mut frames = Vec::with_capacity(points.len());
406 for (index, time) in points {
407 let frame = self.indicator_frame(index, time, trade_data_enabled);
408 self.indicator_host
409 .evaluate_closed_selected(indicator_ids, &frame);
410 frames.push(frame);
411 }
412 frames
413 }
414
415 fn indicator_frame(
416 &self,
417 index: usize,
418 time: exchange::UnixMs,
419 trade_data_enabled: bool,
420 ) -> IndicatorFrame {
421 let base_series = self
422 .base_series()
423 .expect("indicator frame requires a base series");
424 let open_interest = base_series
426 .open_interest()
427 .and_then(|series| series.latest_at_or_before(time));
428 base_series.indicator_frame(index, time, open_interest, trade_data_enabled)
429 }
430
431 fn closed_trade_correction_start(
432 &self,
433 trades: &[Trade],
434 timeframe: Timeframe,
435 ) -> Option<usize> {
436 let earliest_trade_bar = trades
437 .iter()
438 .map(|trade| trade.time.floor_to(timeframe))
439 .min()?;
440 let base_series = self.base_series()?;
441 let latest_closed_time = base_series
442 .points()
443 .rev()
444 .nth(1)
445 .map(|point| point.kline.time)?;
446 if earliest_trade_bar > latest_closed_time {
447 return None;
448 }
449
450 Some(
451 base_series
452 .committed_points()
453 .find(|point| point.kline.time >= earliest_trade_bar)
454 .map_or(0, |point| point.index),
455 )
456 }
457
458 pub fn has_closed_trade_correction(&self, trades: &[Trade], basis: Basis) -> bool {
459 match basis {
460 Basis::Time(timeframe) => self
461 .closed_trade_correction_start(trades, timeframe)
462 .is_some(),
463 Basis::Tick(_) => false,
464 }
465 }
466
467 fn committed_indicator_frames_from(
468 &self,
469 start: usize,
470 trade_data_enabled: bool,
471 ) -> Vec<IndicatorFrame> {
472 self.base_series()
473 .map(|series| {
474 series
475 .committed_points()
476 .skip(start)
477 .map(|point| {
478 self.indicator_frame(point.index, point.kline.time, trade_data_enabled)
479 })
480 .collect()
481 })
482 .unwrap_or_default()
483 }
484
485 pub fn insert_historical_klines(
486 &mut self,
487 ticker_info: TickerInfo,
488 klines: &[Kline],
489 basis: Basis,
490 ) -> Option<KlineUpdate> {
491 let idx = self.series_index.get(&ticker_info).copied()?;
492
493 if klines.is_empty() {
494 return None;
495 }
496 self.series[idx].insert_klines(klines, basis)
497 }
498
499 pub fn update_live_kline(
500 &mut self,
501 ticker_info: &TickerInfo,
502 kline: &Kline,
503 basis: Basis,
504 indicator_ids: &[IndicatorInstanceId],
505 trade_data_enabled: bool,
506 ) -> Option<LiveIndicatorUpdate> {
507 let idx = self.series_index.get(ticker_info).copied()?;
508 let previous_time = self.series[idx]
509 .latest_point()
510 .map(|point| point.kline.time);
511 let next_time = kline.time;
512 let is_base = *ticker_info == self.base_ticker;
513 let starts_new_bar = previous_time.is_some_and(|time| next_time > time);
514 let update = self.series[idx].update_latest_kline(kline, basis)?;
515 let rebuild_scope = (is_base
516 && update.scope.is_full()
517 && previous_time.is_some_and(|time| next_time < time))
518 .then_some(IndicatorUpdateScope::Full);
519 let coverage_frames = if starts_new_bar && is_base {
520 let committed_frame = self.evaluate_closed_previous_frame(
521 previous_time,
522 indicator_ids,
523 trade_data_enabled,
524 );
525 self.preview_indicator_open_for(indicator_ids, trade_data_enabled);
526 committed_frame.into_iter().collect()
527 } else {
528 Vec::new()
529 };
530 Some(LiveIndicatorUpdate {
531 committed_bar: starts_new_bar && is_base,
532 rebuild_scope,
533 closed_trade_correction: false,
534 coverage_frames,
535 index_change: update.index_change,
536 })
537 }
538
539 pub fn insert_live_trades(
540 &mut self,
541 ticker_info: &TickerInfo,
542 trades: &[Trade],
543 basis: Basis,
544 indicator_ids: &[IndicatorInstanceId],
545 trade_data_enabled: bool,
546 ) -> Option<LiveIndicatorUpdate> {
547 if trades.is_empty() {
548 return None;
549 }
550 let index = self.series_index.get(ticker_info).copied()?;
551
552 let is_base = *ticker_info == self.base_ticker;
553 let correction_start = is_base
554 .then(|| match basis {
555 Basis::Time(timeframe) => self.closed_trade_correction_start(trades, timeframe),
556 Basis::Tick(_) => None,
557 })
558 .flatten();
559 let previous_committed_len = if is_base {
560 self.base_series()
561 .map(|series| series.point_count().saturating_sub(1))
562 .unwrap_or_default()
563 } else {
564 0
565 };
566 let scope = match basis {
567 Basis::Tick(_) => self.series[index].insert_tick_trades(trades),
568 Basis::Time(_) => self.series[index].insert_time_trades(trades),
569 };
570 let committed_len = if matches!(basis, Basis::Tick(_)) && is_base {
571 self.base_series()
572 .map(|series| series.point_count().saturating_sub(1))
573 .unwrap_or_default()
574 } else {
575 previous_committed_len
576 };
577 let deferred_append = committed_len.saturating_sub(previous_committed_len) > 1;
578 let newly_committed_frames = if deferred_append {
579 self.committed_indicator_frames_from(previous_committed_len, trade_data_enabled)
580 } else if committed_len > previous_committed_len {
581 let frames = self.evaluate_closed_append_from(
582 previous_committed_len,
583 indicator_ids,
584 trade_data_enabled,
585 );
586 self.preview_indicator_open_for(indicator_ids, trade_data_enabled);
587 frames
588 } else {
589 Vec::new()
590 };
591 let coverage_frames = correction_start
592 .map(|start| self.committed_indicator_frames_from(start, trade_data_enabled))
593 .unwrap_or(newly_committed_frames);
594 Some(LiveIndicatorUpdate {
595 committed_bar: committed_len > previous_committed_len,
596 rebuild_scope: if is_base && scope.is_full() {
597 Some(IndicatorUpdateScope::Full)
598 } else if deferred_append {
599 Some(IndicatorUpdateScope::From(previous_committed_len))
600 } else {
601 None
602 },
603 closed_trade_correction: correction_start.is_some(),
604 coverage_frames,
605 index_change: KlineIndexChange::None,
606 })
607 }
608
609 pub fn insert_historical_trades(
610 &mut self,
611 ticker_info: TickerInfo,
612 trades: &[Trade],
613 is_batches_done: bool,
614 basis: Basis,
615 ) -> bool {
616 let Basis::Time(_) = basis else {
617 return false;
618 };
619
620 let Some(series) = self
621 .series_index
622 .get(&ticker_info)
623 .copied()
624 .and_then(|index| self.series.get_mut(index))
625 else {
626 return false;
627 };
628
629 if !trades.is_empty() {
630 series.insert_time_trades(trades);
631 }
632 !trades.is_empty() || is_batches_done
633 }
634
635 pub fn insert_historical_open_interest(
636 &mut self,
637 ticker_info: TickerInfo,
638 series: &OpenInterestSeries,
639 indicators: &IndicatorInstanceStore,
640 ) -> HistoricalInputUpdate {
641 if ticker_info != self.base_ticker {
642 return HistoricalInputUpdate::Rejected;
643 }
644 let Some(idx) = self.series_index.get(&ticker_info).copied() else {
645 return HistoricalInputUpdate::Rejected;
646 };
647
648 let has_open_interest_consumer = self
649 .indicator_host
650 .has_input_consumer(IndicatorInputSource::OpenInterest, indicators);
651 if series.is_empty() {
652 return HistoricalInputUpdate::NoData {
653 has_consumer: has_open_interest_consumer,
654 };
655 }
656 if !has_open_interest_consumer {
657 return HistoricalInputUpdate::Ignored;
658 }
659 self.series[idx].insert_open_interest(series);
660 HistoricalInputUpdate::Applied
661 }
662
663 pub fn recurring_profile_cache_requirements(
664 &self,
665 studies: impl IntoIterator<Item = KlineStudy>,
666 basis: Basis,
667 ) -> Vec<(Timeframe, PriceStep)> {
668 let tick_size = self.tick_size();
669 let mut recurring_intervals = studies
670 .into_iter()
671 .filter(|study| study.source() == StudyDataSource::Trades)
672 .filter_map(|study| {
673 study
674 .recurring_profile_interval()
675 .filter(|_| study.is_interval_available(basis))
676 .map(|interval| (interval, tick_size))
677 })
678 .collect::<Vec<_>>();
679 recurring_intervals.sort_unstable();
680 recurring_intervals.dedup();
681 recurring_intervals
682 }
683}