1use super::Chart;
4use crate::data::inbox::Inbox;
5use crate::data::merge::{Origin, merge};
6use crate::data::{
7 Datafeed, FeedError, HistoryPage, HistoryRequest, MetricPage, MetricRequest,
8 MetricSubscription, Subscription, SubscriptionId,
9};
10use crate::model::{Bar, BarClock, Instrument, Timeframe};
11use crate::studies::{StudyData, StudyDataRequest};
12use std::collections::BTreeMap;
13use std::sync::{Arc, Mutex};
14
15#[derive(Default)]
16pub(super) struct StudyDataCache {
17 series: Vec<Series>,
18}
19
20struct Series {
21 request: StudyDataRequest,
22 bars: Vec<Bar>,
23 instrument: Option<Instrument>,
24 clock: BarClock,
25 error: Option<FeedError>,
26 stream: Option<Stream>,
27 from_ticks: BTreeMap<chrono::DateTime<chrono::Utc>, Origin>,
28}
29
30struct Stream {
31 feed: Arc<dyn Datafeed>,
32 inbox: Arc<Inbox>,
33 subscription: Option<SubscriptionId>,
34 pending: bool,
35 next_key: u64,
36 history: Option<HistoryRequest>,
38 conversion: Option<String>,
40 cursor: Option<chrono::DateTime<chrono::Utc>>,
42 retry_at: Option<f64>,
43 failures: u32,
44 end: bool,
45 wake: Arc<Mutex<Option<egui::ViewportId>>>,
46}
47
48impl Stream {
49 fn request(&mut self, source: &StudyDataRequest) {
50 self.next_key += 1;
51 self.pending = true;
52 if let Some(history) = &self.history {
53 if let Some(metric) = &source.metric {
54 self.feed.metric_history(
55 MetricRequest {
56 symbol: history.symbol.clone(),
57 metric: metric.clone(),
58 before: history.before,
59 count: history.count,
60 },
61 self.inbox.reply(self.next_key),
62 );
63 } else {
64 self.feed
65 .history(history.clone(), self.inbox.reply(self.next_key));
66 }
67 } else {
68 super::feed::resolve(
69 &*self.feed,
70 &source.symbol,
71 self.conversion.as_deref(),
72 self.inbox.reply(self.next_key),
73 );
74 }
75 }
76
77 fn failed(&mut self, error: &FeedError, now: f64) {
78 if error.is_recoverable() {
79 let delay = super::feed::retry_delay(self.failures, error);
80 self.retry_at = Some(now + delay.as_secs_f64());
81 self.failures = self.failures.saturating_add(1);
82 }
83 }
84}
85
86impl Drop for Stream {
87 fn drop(&mut self) {
88 *self.wake.lock().unwrap_or_else(|error| error.into_inner()) = None;
89 if let Some(id) = self.subscription {
90 self.inbox.unsubscribe(id);
91 self.feed.unsubscribe(id);
92 }
93 }
94}
95
96fn resolved(
97 request: &StudyDataRequest,
98 symbol: &str,
99 timeframe: Option<Timeframe>,
100) -> StudyDataRequest {
101 StudyDataRequest {
102 metric: request.metric.clone(),
103 symbol: if request.symbol.is_empty() {
104 symbol.to_owned()
105 } else {
106 request.symbol.clone()
107 },
108 timeframe: Some(request.timeframe.or(timeframe).unwrap_or_default()),
109 }
110}
111
112fn clock(request: &StudyDataRequest, instrument: Option<&Instrument>, extended: bool) -> BarClock {
113 if request.metric.is_some() {
114 return BarClock::new(Timeframe::Ticks(std::num::NonZeroU64::MIN));
115 }
116 let timeframe = request.timeframe.unwrap_or_default();
117 let mut clock = BarClock::new(timeframe);
118 if let Some(instrument) = instrument {
119 let zone = instrument
120 .time_zone
121 .clone()
122 .unwrap_or_else(crate::model::TimezoneId::utc);
123 if timeframe.duration_ms() < Timeframe::Day1.duration_ms() {
124 clock = clock.with_timezone(zone.clone());
125 }
126 if let Some(session) = &instrument.session {
127 clock = clock.with_session(zone, session.clone(), extended);
128 }
129 }
130 clock
131}
132
133impl StudyDataCache {
134 pub(super) fn snapshot(&self) -> Self {
135 Self {
136 series: self
137 .series
138 .iter()
139 .map(|series| Series {
140 request: series.request.clone(),
141 bars: series.bars.clone(),
142 instrument: series.instrument.clone(),
143 clock: series.clock.clone(),
144 error: series.error.clone(),
145 stream: None,
146 from_ticks: series.from_ticks.clone(),
147 })
148 .collect(),
149 }
150 }
151 pub(super) fn clear_feed(&mut self) {
152 self.series.retain(|series| series.stream.is_none());
153 }
154 pub(super) fn pause(&mut self) {
155 for series in &mut self.series {
156 if let Some(stream) = &mut series.stream {
157 stream.pending = false;
158 stream.retry_at = None;
159 stream.history = None;
160 *stream
161 .wake
162 .lock()
163 .unwrap_or_else(|error| error.into_inner()) = None;
164 if let Some(id) = stream.subscription.take() {
165 stream.inbox.unsubscribe(id);
166 stream.feed.unsubscribe(id);
167 }
168 }
169 }
170 }
171
172 pub(super) fn data(
173 &self,
174 requests: Vec<StudyDataRequest>,
175 symbol: &str,
176 timeframe: Option<Timeframe>,
177 replay: Option<chrono::DateTime<chrono::Utc>>,
178 ) -> Vec<StudyData<'_>> {
179 requests
180 .into_iter()
181 .filter_map(|request| {
182 let key = resolved(&request, symbol, timeframe);
183 let series = self.series.iter().find(|series| series.request == key)?;
184 Some(StudyData {
185 request,
186 bars: match replay {
187 Some(time) => super::replay::completed(&series.bars, &series.clock, time),
188 None => &series.bars,
189 },
190 clock: series.clock.clone(),
191 instrument: series.instrument.as_ref(),
192 timeframe: key.timeframe.unwrap(),
193 })
194 })
195 .collect()
196 }
197}
198
199impl Chart {
200 pub fn study_data_requests(&self) -> Vec<StudyDataRequest> {
204 let mut requests = self.fed_requests();
205 for (_, study) in self.indicators() {
206 for request in study.data_requests() {
207 let request = resolved(&request, self.symbol(), self.timeframe());
208 if !requests.contains(&request) {
209 requests.push(request);
210 }
211 }
212 }
213 #[cfg(feature = "drawings")]
214 for (request, _) in self.profile_histories(&self.drawings) {
215 let request = resolved(&request, self.symbol(), self.timeframe());
216 if !requests.contains(&request) {
217 requests.push(request);
218 }
219 }
220 requests
221 }
222
223 pub fn set_study_data(
227 &mut self,
228 request: StudyDataRequest,
229 bars: Vec<Bar>,
230 instrument: Option<Instrument>,
231 ) {
232 let request = resolved(&request, self.symbol(), self.timeframe());
233 let clock = clock(&request, instrument.as_ref(), self.extended_hours());
234 let (bars, _) = super::series::clean(bars, &clock);
235 self.study_data
236 .series
237 .retain(|series| series.request != request);
238 self.study_data.series.push(Series {
239 request,
240 bars,
241 instrument,
242 clock,
243 error: None,
244 stream: None,
245 from_ticks: BTreeMap::new(),
246 });
247 self.invalidate_study_data();
248 }
249
250 pub fn study_data_error(&self, request: &StudyDataRequest) -> Option<&FeedError> {
254 let request = resolved(request, self.symbol(), self.timeframe());
255 self.study_data
256 .series
257 .iter()
258 .find(|series| series.request == request)?
259 .error
260 .as_ref()
261 }
262
263 pub(super) fn invalidate_study_data(&mut self) {
264 for (at, study) in self.studies.indicators().iter().enumerate() {
265 if !study.data_requests().is_empty() {
266 self.pending_from[at] = Some(0);
267 self.next_refresh[at] = f64::NEG_INFINITY;
268 }
269 }
270 self.element_studies.dirty = true;
271 self.fill_fed_series();
272 }
273
274 fn study_data_reaches(&self) -> Vec<(StudyDataRequest, Option<chrono::DateTime<chrono::Utc>>)> {
279 let full = self.chart_type.is_time_independent();
282 let mut reaches: Vec<(StudyDataRequest, Option<_>)> = self
283 .fed_requests()
284 .into_iter()
285 .map(|request| (request, None))
286 .collect();
287 for (_, study) in self.indicators() {
288 for declared in study.data_requests() {
289 let start = study.data_start(&declared).filter(|_| !full);
290 let request = resolved(&declared, self.symbol(), self.timeframe());
291 match reaches.iter_mut().find(|(known, _)| *known == request) {
292 Some((_, reach)) => *reach = (*reach).min(start),
294 None => reaches.push((request, start)),
295 }
296 }
297 }
298 #[cfg(feature = "drawings")]
299 for (declared, start) in self.profile_histories(&self.drawings) {
300 let request = resolved(&declared, self.symbol(), self.timeframe());
301 match reaches.iter_mut().find(|(known, _)| *known == request) {
302 Some((_, reach)) => *reach = (*reach).min(Some(start)),
303 None => reaches.push((request, Some(start))),
304 }
305 }
306 let first = self.state.data().bars.first().map(|bar| bar.time);
307 for (_, reach) in &mut reaches {
308 *reach = first.map(|first| reach.map_or(first, |start| start.max(first)));
309 }
310 reaches
311 }
312
313 pub(super) fn pump_study_data(&mut self, ctx: &egui::Context, feed: Option<Arc<dyn Datafeed>>) {
314 let requests = self.study_data_reaches();
315 let previous = self.study_data.series.len();
316 self.study_data.series.retain(|series| {
317 requests
318 .iter()
319 .any(|(request, _)| *request == series.request)
320 && series.stream.as_ref().is_none_or(|stream| {
321 feed.as_ref()
322 .is_some_and(|feed| Arc::ptr_eq(feed, &stream.feed))
323 })
324 });
325 let removed = self.study_data.series.len() != previous;
326 if let Some(feed) = &feed {
327 for (request, _) in &requests {
328 if request.symbol.is_empty() {
329 continue;
330 }
331 if self
332 .study_data
333 .series
334 .iter()
335 .any(|series| &series.request == request)
336 {
337 continue;
338 }
339 let inbox = Arc::new(Inbox::default());
340 let wake = Arc::new(Mutex::new(Some(ctx.viewport_id())));
341 let repaint = Arc::clone(&wake);
342 let ctx = ctx.clone();
343 inbox.set_waker(move || {
344 let viewport = *repaint.lock().unwrap_or_else(|error| error.into_inner());
345 if let Some(viewport) = viewport {
346 ctx.request_repaint_of(viewport);
347 }
348 });
349 super::feed::resolve(
350 &**feed,
351 &request.symbol,
352 self.conversion.as_deref(),
353 inbox.reply(0),
354 );
355 self.study_data.series.push(Series {
356 request: request.clone(),
357 bars: Vec::new(),
358 instrument: None,
359 clock: clock(request, None, self.extended_hours()),
360 error: None,
361 from_ticks: BTreeMap::new(),
362 stream: Some(Stream {
363 feed: Arc::clone(feed),
364 inbox,
365 subscription: None,
366 pending: true,
367 next_key: 0,
368 history: None,
369 conversion: self.conversion.clone(),
370 cursor: None,
371 retry_at: None,
372 failures: 0,
373 end: false,
374 wake,
375 }),
376 });
377 }
378 }
379 let extended = self.extended_hours();
380 let adjusted = self.dividend_adjusted();
381 let now = ctx.input(|input| input.time);
382 let count = self.state.data().bars.len().max(300);
383 let mut changed = removed || self.study_data.series.len() != previous;
384 for series in &mut self.study_data.series {
385 let Some(stream) = &mut series.stream else {
386 let clock = clock(&series.request, series.instrument.as_ref(), extended);
387 if clock != series.clock {
388 series.clock = clock;
389 changed = true;
390 }
391 continue;
392 };
393 *stream
394 .wake
395 .lock()
396 .unwrap_or_else(|error| error.into_inner()) = Some(ctx.viewport_id());
397 let drained = stream.inbox.drain();
398 for (key, answer) in drained.answers {
399 if !stream.pending || key != stream.next_key {
400 continue;
401 }
402 stream.pending = false;
403 if stream.history.is_none() {
404 let result = answer
405 .downcast::<Result<Instrument, FeedError>>()
406 .map(|result| *result)
407 .unwrap_or(Err(FeedError::NotSupported));
408 match result {
409 Ok(instrument) => {
410 let tf = series.request.timeframe.unwrap();
411 if series.request.metric.is_none()
412 && instrument
413 .supported_timeframes
414 .as_ref()
415 .is_some_and(|intervals| !intervals.contains(&tf))
416 {
417 series.error = Some(FeedError::NotSupported);
418 continue;
419 }
420 series.clock = clock(&series.request, Some(&instrument), extended);
421 series.instrument = Some(instrument);
422 series.error = None;
423 stream.failures = 0;
424 }
425 Err(error) => {
426 stream.failed(&error, now);
427 series.error = Some(error);
428 }
429 }
430 } else {
431 let result = if series.request.metric.is_some() {
432 answer
433 .downcast::<Result<MetricPage, FeedError>>()
434 .map(|result| {
435 result.map(|page| {
436 HistoryPage::new(
437 page.points.into_iter().map(Bar::from).collect(),
438 page.end_of_history,
439 )
440 })
441 })
442 .unwrap_or(Err(FeedError::NotSupported))
443 } else {
444 answer
445 .downcast::<Result<HistoryPage, FeedError>>()
446 .map(|result| *result)
447 .unwrap_or(Err(FeedError::NotSupported))
448 };
449 match result {
450 Ok(mut page) => {
451 let previous = stream.history.take().and_then(|request| request.before);
452 page.bars.retain(|bar| {
453 previous
454 .is_none_or(|before| series.clock.bar_open(bar.time) < before)
455 });
456 series.error = None;
457 stream.failures = 0;
458 let empty = page.bars.is_empty();
459 let continuation = page.continue_before.filter(|_| empty);
460 stream.cursor = continuation;
461 stream.end =
462 page.end_of_history || (empty && page.continue_before.is_none());
463 if continuation.is_some_and(|at| previous.is_some_and(|b| at >= b)) {
464 stream.cursor = None;
465 stream.end = true;
466 series.error = Some(FeedError::InvalidData(
467 "history continuation does not lie before the request".into(),
468 ));
469 }
470 let incoming = page
471 .bars
472 .into_iter()
473 .map(|bar| (bar, Origin::History))
474 .collect();
475 merge(
476 &mut series.bars,
477 &mut series.from_ticks,
478 &series.clock,
479 incoming,
480 );
481 if previous.is_some()
482 && continuation.is_none()
483 && series.bars.first().map(|bar| bar.time) >= previous
484 {
485 stream.end = true;
486 }
487 changed = true;
488 }
489 Err(error) => {
490 stream.failed(&error, now);
491 series.error = Some(error);
492 }
493 }
494 }
495 }
496 for (_, updates) in drained.updates {
497 if updates.reset {
498 series.bars.clear();
499 series.from_ticks.clear();
500 series.error = None;
501 stream.pending = false;
502 stream.history = None;
503 stream.retry_at = None;
504 stream.failures = 0;
505 stream.end = false;
506 changed = true;
507 }
508 if !updates.bars.is_empty() {
509 merge(
510 &mut series.bars,
511 &mut series.from_ticks,
512 &series.clock,
513 updates.bars.into_values().collect(),
514 );
515 changed = true;
516 }
517 if updates.error.is_some() {
518 series.error = updates.error;
519 }
520 }
521 if stream.subscription.is_none()
522 && series.instrument.is_some()
523 && series.error.is_none()
524 {
525 let tf = series.request.timeframe.unwrap();
526 let sink = stream.inbox.subscribe(
527 series.request.symbol.clone(),
528 series.clock.clone(),
529 super::feed::TRADE_CAPACITY,
530 series.instrument.as_ref().and_then(|i| i.tick_size),
531 );
532 stream.subscription = Some(sink.id);
533 if let Some(metric) = &series.request.metric {
534 stream.feed.subscribe_metric(MetricSubscription {
535 id: sink.id,
536 symbol: series.request.symbol.clone(),
537 metric: metric.clone(),
538 sink,
539 });
540 } else {
541 stream.feed.subscribe(Subscription {
542 id: sink.id,
543 symbol: series.request.symbol.clone(),
544 timeframe: tf,
545 clock: series.clock.clone(),
546 extended_hours: extended,
547 conversion: stream.conversion.clone(),
548 sink,
549 });
550 }
551 }
552 if let Some(deadline) = stream.retry_at {
553 if now >= deadline {
554 stream.retry_at = None;
555 stream.request(&series.request);
556 } else {
557 ctx.request_repaint_after(
558 std::time::Duration::try_from_secs_f64(deadline - now)
559 .unwrap_or(std::time::Duration::MAX),
560 );
561 }
562 }
563 let reach = requests
564 .iter()
565 .find(|(request, _)| *request == series.request)
566 .and_then(|(_, reach)| *reach);
567 if !stream.pending
568 && stream.retry_at.is_none()
569 && !stream.end
570 && series.error.is_none()
571 && series.instrument.is_some()
572 && (series.bars.is_empty()
573 || reach
574 .zip(series.bars.first())
575 .is_some_and(|(reach, bar)| reach < bar.time))
576 {
577 stream.history = Some(HistoryRequest {
578 symbol: series.request.symbol.clone(),
579 timeframe: series.request.timeframe.unwrap(),
580 clock: series.clock.clone(),
581 before: series
582 .bars
583 .first()
584 .map(|bar| bar.time)
585 .into_iter()
586 .chain(stream.cursor)
587 .min(),
588 since: None,
589 count,
590 extended_hours: extended,
591 adjusted,
592 conversion: stream.conversion.clone(),
593 });
594 stream.request(&series.request);
595 }
596 }
597 if changed {
598 self.invalidate_study_data();
599 }
600 }
601}
602
603#[cfg(all(test, feature = "studies"))]
604mod tests {
605 use super::*;
606 use crate::data::{Reply, Sink};
607 use crate::studies::{InputValue, SMA};
608 use crate::widget::{ChartBuilder, test_frames::shapes};
609 use chrono::{DateTime, Duration};
610
611 #[derive(Default)]
612 struct Feed {
613 sinks: Mutex<Vec<(String, Sink)>>,
614 correction: Mutex<f64>,
615 metrics: Mutex<Vec<MetricSubscription>>,
616 }
617 impl Datafeed for Feed {
618 fn history(&self, request: HistoryRequest, reply: Reply<HistoryPage>) {
619 let (prices, minutes) = match request.symbol.as_str() {
620 "REMOTE" => (vec![10.0, 14.0, 20.0], 60),
621 "OTHER" => (
622 vec![100.0, 200.0, 300.0 + *self.correction.lock().unwrap()],
623 60,
624 ),
625 _ => (vec![1.0; 5], 30),
626 };
627 reply.send(Ok(HistoryPage::new(bars(&prices, minutes), true)));
628 }
629 fn metric_history(&self, _: MetricRequest, reply: Reply<MetricPage>) {
630 reply.send(Ok(MetricPage {
631 points: [(15, 10.0), (75, 20.0)]
632 .into_iter()
633 .map(|(minute, value)| crate::data::MetricPoint {
634 time: DateTime::UNIX_EPOCH + Duration::minutes(minute),
635 value,
636 })
637 .collect(),
638 end_of_history: true,
639 }));
640 }
641 fn subscribe_metric(&self, subscription: MetricSubscription) {
642 self.metrics.lock().unwrap().push(subscription);
643 }
644 fn subscribe(&self, subscription: Subscription) {
645 self.sinks
646 .lock()
647 .unwrap()
648 .push((subscription.symbol, subscription.sink));
649 }
650 }
651 fn bars(prices: &[f64], minutes: i64) -> Vec<Bar> {
652 prices
653 .iter()
654 .enumerate()
655 .map(|(at, &price)| {
656 Bar::new(
657 DateTime::UNIX_EPOCH + Duration::minutes(at as i64 * minutes),
658 price,
659 price,
660 price,
661 price,
662 1.0,
663 )
664 })
665 .collect()
666 }
667
668 struct Counting;
670
671 thread_local! {
672 static ALLOCATIONS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
673 }
674
675 #[global_allocator]
676 static ALLOCATOR: Counting = Counting;
677
678 unsafe impl std::alloc::GlobalAlloc for Counting {
680 unsafe fn alloc(&self, layout: std::alloc::Layout) -> *mut u8 {
681 let _ = ALLOCATIONS.try_with(|count| count.set(count.get() + 1));
682 unsafe { std::alloc::System.alloc(layout) }
684 }
685
686 unsafe fn dealloc(&self, ptr: *mut u8, layout: std::alloc::Layout) {
687 unsafe { std::alloc::System.dealloc(ptr, layout) }
689 }
690
691 unsafe fn realloc(
692 &self,
693 ptr: *mut u8,
694 layout: std::alloc::Layout,
695 new_size: usize,
696 ) -> *mut u8 {
697 let _ = ALLOCATIONS.try_with(|count| count.set(count.get() + 1));
698 unsafe { std::alloc::System.realloc(ptr, layout, new_size) }
700 }
701 }
702
703 fn allocations<T>(run: impl FnOnce() -> T) -> (usize, T) {
705 let before = ALLOCATIONS.with(std::cell::Cell::get);
706 let result = run();
707 (ALLOCATIONS.with(std::cell::Cell::get) - before, result)
708 }
709
710 #[derive(Default)]
711 struct RetryFeed {
712 resolves: Mutex<Vec<crate::data::Reply<Instrument>>>,
713 histories: Mutex<Vec<(HistoryRequest, crate::data::Reply<HistoryPage>)>>,
714 metrics: Mutex<Vec<(MetricRequest, crate::data::Reply<MetricPage>)>>,
715 sinks: Mutex<Vec<Sink>>,
716 }
717 impl Datafeed for RetryFeed {
718 fn resolve(&self, _: &str, reply: Reply<Instrument>) {
719 self.resolves.lock().unwrap().push(reply);
720 }
721 fn history(&self, request: HistoryRequest, reply: Reply<HistoryPage>) {
722 self.histories.lock().unwrap().push((request, reply));
723 }
724 fn metric_history(&self, request: MetricRequest, reply: Reply<MetricPage>) {
725 self.metrics.lock().unwrap().push((request, reply));
726 }
727 fn subscribe(&self, subscription: Subscription) {
728 self.sinks.lock().unwrap().push(subscription.sink);
729 }
730 fn subscribe_metric(&self, subscription: MetricSubscription) {
731 self.sinks.lock().unwrap().push(subscription.sink);
732 }
733 }
734 impl RetryFeed {
735 fn pending(&self, metric: bool) -> usize {
736 if metric {
737 self.metrics.lock().unwrap().len()
738 } else {
739 self.histories.lock().unwrap().len()
740 }
741 }
742 fn answer(
743 &self,
744 metric: bool,
745 result: Result<HistoryPage, FeedError>,
746 ) -> (Option<DateTime<chrono::Utc>>, usize) {
747 if metric {
748 let (request, reply) = self.metrics.lock().unwrap().remove(0);
749 reply.send(result.map(|page| {
750 MetricPage {
751 points: page
752 .bars
753 .into_iter()
754 .map(|bar| crate::data::MetricPoint {
755 time: bar.time,
756 value: bar.close,
757 })
758 .collect(),
759 end_of_history: page.end_of_history,
760 }
761 }));
762 (request.before, request.count)
763 } else {
764 let (request, reply) = self.histories.lock().unwrap().remove(0);
765 reply.send(result);
766 (request.before, request.count)
767 }
768 }
769 }
770
771 #[test]
772 fn study_sources_retry_original_requests_and_ignore_superseded_replies() {
773 for metric in [false, true] {
774 for (error, delay) in [
775 (FeedError::Network("offline".into()), 1.0),
776 (
777 FeedError::RateLimited {
778 retry_after: Some(std::time::Duration::from_secs(7)),
779 },
780 7.0,
781 ),
782 (FeedError::Unanswered, 1.0),
783 (FeedError::NotSupported, 0.0),
784 ] {
785 let feed = Arc::new(RetryFeed::default());
786 let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 5], 30)));
787 chart.set_symbol("REMOTE");
788 if metric {
789 chart.add_indicator(Box::new(crate::studies::MetricStudy::new(
790 crate::data::MetricSeries {
791 id: "oi".into(),
792 name: "Open interest".into(),
793 unit: "contracts".into(),
794 format: crate::scales::StudyFormat::Volume,
795 },
796 )));
797 } else {
798 let id = chart.add_indicator(Box::new(SMA::new(1)));
799 chart
800 .indicator_mut(id)
801 .unwrap()
802 .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
803 .unwrap();
804 }
805 let ctx = egui::Context::default();
806 let pump = |chart: &mut Chart, time| {
807 ctx.run_ui(
808 egui::RawInput {
809 time: Some(time),
810 ..Default::default()
811 },
812 |_| {},
813 )
814 .drop_without_applying_deltas();
815 chart.pump_study_data(&ctx, Some(feed.clone()));
816 };
817 pump(&mut chart, 0.0);
818 feed.resolves
819 .lock()
820 .unwrap()
821 .pop()
822 .unwrap()
823 .send(Err(error.clone()));
824 pump(&mut chart, 0.25);
825 let request = chart.study_data_requests().pop().unwrap();
826 assert_eq!(chart.study_data_error(&request), Some(&error));
827 pump(&mut chart, 0.25 + delay - 0.01);
828 assert!(feed.resolves.lock().unwrap().is_empty());
829 pump(&mut chart, 0.25 + delay);
830 if !error.is_recoverable() {
831 pump(&mut chart, 120.0);
832 assert!(feed.resolves.lock().unwrap().is_empty());
833 continue;
834 }
835 assert_eq!(feed.resolves.lock().unwrap().len(), 1, "{error:?}");
836 feed.resolves
837 .lock()
838 .unwrap()
839 .pop()
840 .unwrap()
841 .send(Ok(Instrument::new("REMOTE")));
842 let now = 0.5 + delay;
843 pump(&mut chart, now);
844 assert_eq!(chart.study_data_error(&request), None);
845 let mut newest = bars(&[20.0], 60);
846 newest[0].time += Duration::hours(2);
847 assert_eq!(
848 feed.answer(metric, Ok(HistoryPage::new(newest, false))).0,
849 None
850 );
851 pump(&mut chart, now + 0.25);
852 let failed = feed.answer(metric, Err(error.clone()));
853 pump(&mut chart, now + 0.5);
854 feed.sinks.lock().unwrap()[0].push_bar(Bar::new(
855 DateTime::UNIX_EPOCH + Duration::hours(1),
856 55.0,
857 55.0,
858 55.0,
859 55.0,
860 1.0,
861 ));
862 pump(&mut chart, now + 0.5 + delay - 0.01);
863 assert_eq!(feed.pending(metric), 0);
864 pump(&mut chart, now + 0.5 + delay);
865 assert_eq!(feed.pending(metric), 1);
866 let retried = feed.answer(
867 metric,
868 Ok(HistoryPage::new(bars(&[12.0, 55.0, 99.0, 100.0], 60), true)),
869 );
870 assert_eq!(
871 retried, failed,
872 "retry must keep its boundary and count after a live prepend"
873 );
874 assert_eq!(retried.0, Some(DateTime::UNIX_EPOCH + Duration::hours(2)));
875 pump(&mut chart, now + 0.75 + delay);
876 assert_eq!(chart.study_data_error(&request), None);
877 assert_eq!(
878 chart.study_data.series[0]
879 .bars
880 .iter()
881 .map(|bar| bar.close)
882 .collect::<Vec<_>>(),
883 [12.0, 55.0, 20.0]
884 );
885 for time in [now + 10.0, now + 11.0] {
887 feed.sinks.lock().unwrap()[0].reset_replay(DateTime::UNIX_EPOCH);
888 pump(&mut chart, time);
889 }
890 feed.answer(metric, Err(FeedError::Network("old page".into())));
891 pump(&mut chart, now + 12.0);
892 assert_eq!(chart.study_data_error(&request), None);
893 feed.answer(metric, Ok(HistoryPage::new(bars(&[99.0], 60), true)));
894 pump(&mut chart, now + 13.0);
895 assert_eq!(chart.study_data_error(&request), None);
896 assert_eq!(chart.study_data.series[0].bars[0].close, 99.0);
897 assert_eq!(feed.pending(metric), 0);
898 }
899 }
900 }
901
902 #[test]
903 fn study_retries_back_off_and_cancel_with_their_source() {
904 for action in ["pause", "host data", "remove", "replace"] {
905 let feed = Arc::new(RetryFeed::default());
906 let mut chart = Chart::new(crate::model::BarData::new());
907 let id = chart.add_indicator(Box::new(SMA::new(1)));
908 chart
909 .indicator_mut(id)
910 .unwrap()
911 .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
912 .unwrap();
913 let ctx = egui::Context::default();
914 let pump = |chart: &mut Chart, time| {
915 ctx.run_ui(
916 egui::RawInput {
917 time: Some(time),
918 ..Default::default()
919 },
920 |_| {},
921 )
922 .drop_without_applying_deltas();
923 chart.pump_study_data(&ctx, Some(feed.clone()));
924 };
925 pump(&mut chart, 0.0);
926 feed.resolves
927 .lock()
928 .unwrap()
929 .pop()
930 .unwrap()
931 .send(Err(FeedError::Network("offline".into())));
932 pump(&mut chart, 0.1);
933 pump(&mut chart, 1.1);
934 feed.resolves
935 .lock()
936 .unwrap()
937 .pop()
938 .unwrap()
939 .send(Err(FeedError::Network("still offline".into())));
940 pump(&mut chart, 1.2);
941 pump(&mut chart, 2.2);
942 assert!(
943 feed.resolves.lock().unwrap().is_empty(),
944 "second failure waits two seconds"
945 );
946 match action {
947 "pause" => {
948 chart.set_feed_paused(true);
949 }
950 "host data" => {
951 let request = chart.study_data_requests().pop().unwrap();
952 chart.set_study_data(request, bars(&[99.0], 60), None);
953 }
954 "remove" => {
955 chart.remove_series(id);
956 }
957 "replace" => {
958 chart
959 .indicator_mut(id)
960 .unwrap()
961 .set_input("calculation_symbol", InputValue::Symbol("OTHER".into()))
962 .unwrap();
963 }
964 _ => unreachable!(),
965 }
966 pump(&mut chart, 60.0);
967 assert_eq!(
968 feed.resolves.lock().unwrap().len(),
969 usize::from(action == "replace"),
970 "{action}"
971 );
972 }
973 }
974
975 #[test]
976 fn replay_reset_discards_future_study_bars_and_tick_origins() {
977 for replacement in [false, true] {
978 let feed = Arc::new(Feed::default());
979 let mut chart = ChartBuilder::new()
980 .with_feed(feed.clone())
981 .with_symbol("MAIN")
982 .with_timeframe(Timeframe::Hour1)
983 .build();
984 let id = chart.add_indicator(Box::new(SMA::new(1)));
985 chart
986 .indicator_mut(id)
987 .unwrap()
988 .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
989 .unwrap();
990 let ctx = egui::Context::default();
991 for frame in 0..8 {
992 shapes(&ctx, &mut chart, frame as f64, Vec::new());
993 }
994 let sink = feed
995 .sinks
996 .lock()
997 .unwrap()
998 .iter()
999 .find(|(symbol, _)| symbol == "REMOTE")
1000 .unwrap()
1001 .1
1002 .clone();
1003 sink.set_server_time(DateTime::UNIX_EPOCH + Duration::hours(4));
1004 sink.push_tick(crate::model::Tick::new(
1005 DateTime::UNIX_EPOCH + Duration::hours(3),
1006 30.0,
1007 1.0,
1008 None,
1009 ));
1010 chart.pump_study_data(&ctx, Some(feed.clone()));
1011 assert!(!chart.study_data.series[0].from_ticks.is_empty());
1012 sink.reset_replay(DateTime::UNIX_EPOCH);
1013 if replacement {
1014 sink.push_tick(crate::model::Tick::new(
1015 DateTime::UNIX_EPOCH,
1016 99.0,
1017 1.0,
1018 None,
1019 ));
1020 }
1021 chart.pump_study_data(&ctx, Some(feed.clone()));
1022 let series = &chart.study_data.series[0];
1023 assert_eq!(series.bars.len(), usize::from(replacement));
1024 assert!(
1025 series
1026 .from_ticks
1027 .keys()
1028 .all(|time| *time == DateTime::UNIX_EPOCH)
1029 );
1030 if replacement {
1031 assert_eq!(series.bars[0].close, 99.0);
1032 }
1033 assert_eq!(chart.pending_from[0], Some(0));
1034 }
1035 }
1036
1037 #[test]
1040 fn metric_history_and_live_points_refresh_the_pane_and_its_consumers() {
1041 use crate::data::{MetricPoint, MetricSeries};
1042 use crate::scales::StudyFormat;
1043 use crate::studies::{MetricStudy, StudySource};
1044 let feed = Arc::new(Feed::default());
1045 let mut chart = ChartBuilder::new()
1046 .with_feed(feed.clone())
1047 .with_symbol("MAIN")
1048 .with_timeframe(Timeframe::Min30)
1049 .build();
1050 let metric = chart.add_indicator(Box::new(MetricStudy::new(MetricSeries {
1051 id: "oi".into(),
1052 name: "Open interest".into(),
1053 unit: "contracts".into(),
1054 format: StudyFormat::Volume,
1055 })));
1056 let average = chart.add_indicator(Box::new(SMA::new(2)));
1057 chart
1058 .set_indicator_source(
1059 average,
1060 StudySource::Study {
1061 id: metric.0,
1062 plot: 0,
1063 },
1064 )
1065 .unwrap();
1066 let ctx = egui::Context::default();
1067 let settle = |chart: &mut Chart| {
1068 for frame in 0..12 {
1069 shapes(&ctx, chart, frame as f64, Vec::new());
1070 }
1071 };
1072 let output = |chart: &Chart, id| {
1073 let study = chart.indicators().find(|(key, _)| *key == id).unwrap().1;
1074 (0..5)
1075 .map(|at| study.values().value(0, at))
1076 .collect::<Vec<_>>()
1077 };
1078 settle(&mut chart);
1079 assert_eq!(
1080 output(&chart, metric),
1081 [None, Some(10.0), Some(10.0), Some(20.0), Some(20.0)]
1082 );
1083 assert_eq!(
1084 output(&chart, average),
1085 [None, None, Some(10.0), Some(15.0), Some(20.0)]
1086 );
1087 let sink = feed.metrics.lock().unwrap()[0].sink.clone();
1088 for (minutes, value) in [(15, 14.0), (110, 30.0)] {
1089 sink.push_metric(MetricPoint {
1090 time: DateTime::UNIX_EPOCH + Duration::minutes(minutes),
1091 value,
1092 });
1093 }
1094 for frame in 0..60 {
1095 let mut last = bars(&[1.0; 5], 30).pop().unwrap();
1096 last.close = 1.0 + frame as f64 * 0.001;
1097 last.high = last.close;
1098 chart.update(super::super::SeriesId::MAIN, last);
1099 shapes(&ctx, &mut chart, 12.0 + frame as f64 / 60.0, Vec::new());
1100 }
1101 assert_eq!(
1102 output(&chart, metric),
1103 [None, Some(14.0), Some(14.0), Some(20.0), Some(30.0)]
1104 );
1105 assert_eq!(
1106 output(&chart, average),
1107 [None, None, Some(14.0), Some(17.0), Some(25.0)]
1108 );
1109 chart.set_symbol("OTHER");
1110 settle(&mut chart);
1111 assert!(!sink.is_active());
1112 drop(chart);
1113 assert!(
1114 feed.metrics
1115 .lock()
1116 .unwrap()
1117 .iter()
1118 .all(|sub| !sub.sink.is_active())
1119 );
1120 }
1121
1122 #[test]
1123 fn requested_intervals_align_confirmed_and_live_values_and_replace_subscriptions() {
1124 let feed = Arc::new(Feed::default());
1125 let mut chart = ChartBuilder::new()
1126 .with_feed(feed.clone())
1127 .with_symbol("MAIN")
1128 .with_timeframe(Timeframe::Min30)
1129 .build();
1130 let id = chart.add_indicator(Box::new(SMA::new(2)));
1131 let study = chart.indicator_mut(id).unwrap();
1132 study
1133 .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
1134 .unwrap();
1135 study
1136 .set_input(
1137 "calculation_timeframe",
1138 InputValue::Timeframe(Some(Timeframe::Hour1)),
1139 )
1140 .unwrap();
1141 let ctx = egui::Context::default();
1142 let settle = |chart: &mut Chart| {
1143 for frame in 0..10 {
1144 shapes(&ctx, chart, frame as f64, Vec::new());
1145 }
1146 };
1147 settle(&mut chart);
1148 let output = |chart: &Chart| {
1149 (0..5)
1150 .map(|at| {
1151 chart
1152 .indicators()
1153 .find(|(candidate, _)| *candidate == id)
1154 .unwrap()
1155 .1
1156 .values()
1157 .value(0, at)
1158 })
1159 .collect::<Vec<_>>()
1160 };
1161 assert_eq!(output(&chart), [None, None, None, Some(12.0), Some(12.0)]);
1163 chart.set_data(super::super::SeriesId::MAIN, bars(&[1.0; 6], 30));
1166 settle(&mut chart);
1167 assert_eq!(
1168 chart
1169 .indicators()
1170 .find(|(candidate, _)| *candidate == id)
1171 .unwrap()
1172 .1
1173 .values()
1174 .value(0, 5),
1175 Some(12.0)
1176 );
1177 chart.set_data(super::super::SeriesId::MAIN, bars(&[1.0; 5], 30));
1178 settle(&mut chart);
1179 chart
1180 .indicator_mut(id)
1181 .unwrap()
1182 .set_input("wait_for_close", InputValue::Toggle(false))
1183 .unwrap();
1184 settle(&mut chart);
1185 assert_eq!(output(&chart), [None, None, None, Some(12.0), Some(17.0)]);
1186 let sink = feed
1187 .sinks
1188 .lock()
1189 .unwrap()
1190 .iter()
1191 .find(|(symbol, _)| symbol == "REMOTE")
1192 .unwrap()
1193 .1
1194 .clone();
1195 sink.push_bar(bars(&[10.0, 14.0, 24.0], 60).pop().unwrap());
1196 settle(&mut chart);
1197 assert_eq!(output(&chart)[4], Some(19.0));
1198 chart
1199 .indicator_mut(id)
1200 .unwrap()
1201 .set_input("calculation_symbol", InputValue::Symbol("OTHER".into()))
1202 .unwrap();
1203 settle(&mut chart);
1204 assert!(!sink.is_active());
1205 assert_eq!(output(&chart)[4], Some(250.0));
1206 let other_sink = feed
1207 .sinks
1208 .lock()
1209 .unwrap()
1210 .iter()
1211 .rev()
1212 .find(|(symbol, _)| symbol == "OTHER")
1213 .unwrap()
1214 .1
1215 .clone();
1216 chart.set_extended_hours(!chart.extended_hours());
1217 settle(&mut chart);
1218 assert!(!other_sink.is_active());
1219 assert_eq!(output(&chart)[4], Some(250.0));
1220 chart.set_feed_paused(true);
1221 *feed.correction.lock().unwrap() = 100.0;
1222 chart.set_feed_paused(false);
1223 settle(&mut chart);
1224 assert_eq!(
1225 output(&chart)[4],
1226 Some(300.0),
1227 "resuming reloads missed bars"
1228 );
1229 let request = chart.study_data_requests().pop().unwrap();
1230 chart.set_study_data(request, bars(&[2.0, 4.0, 8.0], 60), None);
1231 settle(&mut chart);
1232 assert_eq!(output(&chart)[4], Some(6.0));
1233 chart
1234 .indicator_mut(id)
1235 .unwrap()
1236 .set_input("wait_for_close", InputValue::Toggle(true))
1237 .unwrap();
1238 for (name, minutes, stamp, missing, expected) in [
1239 (
1240 "stamped inside hourly slots",
1241 60,
1242 10,
1243 None,
1244 [None, None, None, Some(3.0), Some(3.0)],
1245 ),
1246 ("missing hourly slot", 60, 0, Some(1), [None; 5]),
1247 (
1248 "missing same-interval slot",
1249 30,
1250 0,
1251 Some(1),
1252 [None, None, Some(5.0), None, None],
1253 ),
1254 ] {
1255 let timeframe = if minutes == 60 {
1256 Timeframe::Hour1
1257 } else {
1258 Timeframe::Min30
1259 };
1260 chart
1261 .indicator_mut(id)
1262 .unwrap()
1263 .set_input(
1264 "calculation_timeframe",
1265 InputValue::Timeframe(Some(timeframe)),
1266 )
1267 .unwrap();
1268 let source = bars(&[2.0, 4.0, 8.0], minutes)
1269 .into_iter()
1270 .enumerate()
1271 .filter(|(at, _)| Some(*at) != missing)
1272 .map(|(_, mut bar)| {
1273 bar.time += Duration::minutes(stamp);
1274 bar
1275 })
1276 .collect();
1277 let request = chart.study_data_requests().pop().unwrap();
1278 chart.set_study_data(request, source, None);
1279 settle(&mut chart);
1280 assert_eq!(output(&chart), expected, "{name}");
1281 }
1282 }
1283
1284 #[test]
1289 fn replay_hides_the_future_of_other_histories() {
1290 let hourly = bars(&[1.0; 3], 60);
1291 let mut chart = Chart::new(crate::model::BarData::from_bars(hourly));
1292 chart.set_symbol("LOCAL");
1293 chart.set_timeframe(Timeframe::Hour1);
1294 let id = chart.add_indicator(Box::new(SMA::new(1)));
1295 chart
1296 .indicator_mut(id)
1297 .unwrap()
1298 .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
1299 .unwrap();
1300 let request = chart.study_data_requests().pop().unwrap();
1301 chart.set_study_data(request, bars(&[10.0, 14.0, 24.0], 60), None);
1302 let ctx = egui::Context::default();
1303 let mut frame = 0.0;
1304 let mut values = |chart: &mut Chart| {
1305 for _ in 0..4 {
1306 frame += 1.0;
1307 shapes(&ctx, chart, frame, Vec::new());
1308 }
1309 let study = chart.indicators().next().unwrap().1;
1310 let shown: Vec<_> = (0..study.values().len())
1311 .map(|at| study.values().value(0, at))
1312 .collect();
1313 #[cfg(feature = "csv")]
1314 {
1315 let exported = chart.export_csv().unwrap();
1316 let mut reader = csv::Reader::from_reader(exported.as_bytes());
1317 let column = reader
1318 .headers()
1319 .unwrap()
1320 .iter()
1321 .position(|name| name == "SMA: SMA(1)")
1322 .unwrap();
1323 let exported: Vec<Option<f64>> = reader
1324 .records()
1325 .map(|row| row.unwrap()[column].parse().ok())
1326 .collect();
1327 assert_eq!(exported, shown, "export uses the same replay cutoff");
1328 }
1329 shown
1330 };
1331 let minute = |m: i64| DateTime::UNIX_EPOCH + Duration::minutes(m);
1332 chart.set_replay_time(Some(minute(90)));
1333 assert!(chart.set_replay_source(
1334 bars(&[1.0; 180], 1),
1335 crate::model::BarClock::new(Timeframe::Min1)
1336 ));
1337 assert_eq!(values(&mut chart), [Some(10.0), None]);
1339 chart.set_replay_time(Some(minute(150)));
1340 assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), None]);
1341 chart.set_replay_time(Some(minute(90)));
1342 assert_eq!(values(&mut chart), [Some(10.0), None]);
1343 chart.set_replay_time(None);
1344 assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), Some(24.0)]);
1345
1346 let driver = Arc::new(
1347 crate::data::ReplayDriver::new(
1348 (0..180)
1349 .map(|m| {
1350 crate::data::ReplayEvent::new(
1351 chart.symbol(),
1352 minute(m),
1353 m as u64,
1354 crate::data::MarketEvent::Trade(crate::model::Tick::new(
1355 minute(m),
1356 1.0,
1357 1.0,
1358 None,
1359 )),
1360 )
1361 })
1362 .collect(),
1363 vec![crate::model::Instrument::new(chart.symbol())],
1364 )
1365 .unwrap(),
1366 );
1367 driver.seek(minute(90));
1368 chart.set_market_replay(Some(driver.clone()));
1369 assert_eq!(values(&mut chart), [Some(10.0), None]);
1370 driver.seek(minute(150));
1371 assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), None]);
1372 driver.seek(minute(90));
1373 assert_eq!(values(&mut chart), [Some(10.0), None]);
1374 chart.set_market_replay(None);
1375 assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), Some(24.0)]);
1376 }
1377
1378 #[test]
1382 fn finer_profile_history_stops_at_the_profile_span() {
1383 use crate::model::ChartType;
1384 let feed = Arc::new(RetryFeed::default());
1385 let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 10], 60)));
1386 chart.set_symbol("MAIN");
1387 chart.set_timeframe(Timeframe::Hour1);
1388 let mut profile: Box<dyn crate::studies::Indicator> =
1389 Box::new(crate::studies::VolProfileVisible::new(10));
1390 profile
1391 .set_input(
1392 "source_interval",
1393 InputValue::Timeframe(Some(Timeframe::Min30)),
1394 )
1395 .unwrap();
1396 let id = chart.add_indicator(profile);
1397 let ctx = egui::Context::default();
1398 let fine = bars(&[1.0; 20], 30);
1400 let hour = |hour: i64| Some(DateTime::UNIX_EPOCH + Duration::hours(hour));
1401 for (chart_type, view, expected) in [
1402 (ChartType::Candles, 8..10, vec![None, hour(9)]),
1403 (ChartType::Candles, 5..10, vec![hour(8), hour(7), hour(6)]),
1404 (ChartType::Candles, 9..10, vec![]),
1405 (
1406 ChartType::Renko,
1407 9..10,
1408 vec![hour(5), hour(4), hour(3), hour(2), hour(1)],
1409 ),
1410 ] {
1411 chart.set_chart_type(chart_type);
1412 chart
1413 .indicator_mut(id)
1414 .unwrap()
1415 .set_visible_range(view.clone());
1416 let mut befores = Vec::new();
1417 loop {
1418 ctx.run_ui(egui::RawInput::default(), |_| {})
1419 .drop_without_applying_deltas();
1420 chart.pump_study_data(&ctx, Some(feed.clone()));
1421 let resolve = feed.resolves.lock().unwrap().pop();
1422 if let Some(reply) = resolve {
1423 reply.send(Ok(Instrument::new("MAIN")));
1424 continue;
1425 }
1426 let Some((request, reply)) = feed.histories.lock().unwrap().pop() else {
1427 break;
1428 };
1429 befores.push(request.before);
1430 let end = fine
1431 .partition_point(|bar| request.before.is_none_or(|before| bar.time < before));
1432 reply.send(Ok(HistoryPage::new(
1433 fine[end.saturating_sub(2)..end].to_vec(),
1434 end <= 2,
1435 )));
1436 }
1437 assert_eq!(befores, expected, "{chart_type:?} {view:?}");
1438 }
1439 }
1440
1441 #[test]
1448 fn waiting_profiles_keep_their_finer_history() {
1449 use crate::model::ProfileResolution;
1450 use crate::studies::{Indicator, VolProfileSession, VolProfileVisible};
1451 let hour = |hour: i64| DateTime::UNIX_EPOCH + Duration::hours(hour);
1452 let hourly = bars(&[1.0; 400], 60);
1453 let suggested = |intervals: &[(Timeframe, Option<i64>)]| {
1454 Instrument::new("MAIN").with_profile_resolutions(intervals.iter().map(
1455 |&(timeframe, days)| {
1456 ProfileResolution::new(timeframe, days.map(chrono::TimeDelta::days))
1457 },
1458 ))
1459 };
1460 let explicit = |mut study: Box<dyn Indicator>, timeframe| {
1461 study
1462 .set_input("source_interval", InputValue::Timeframe(Some(timeframe)))
1463 .unwrap();
1464 study
1465 };
1466 let session = || Box::new(VolProfileSession::default()) as Box<dyn Indicator>;
1467 let visible = || Box::new(VolProfileVisible::default()) as Box<dyn Indicator>;
1468 let finer = [(Timeframe::Min15, Some(1)), (Timeframe::Min30, None)];
1469 let half = [(Timeframe::Min30, None)];
1470 for (name, instrument, studies, main_page, last) in [
1471 (
1472 "session on 15 minutes, visible range on 30, as suggested",
1473 suggested(&finer),
1474 [session(), visible()],
1475 400,
1476 22,
1477 ),
1478 (
1479 "older chart bars arrive after the half-hour history",
1480 suggested(&half),
1481 [Box::new(SMA::new(5)) as Box<dyn Indicator>, visible()],
1482 200,
1483 22,
1484 ),
1485 (
1486 "both on suggested 30 minutes",
1487 suggested(&half),
1488 [session(), visible()],
1489 400,
1490 2,
1491 ),
1492 (
1493 "both on explicit 30 minutes",
1494 suggested(&[]),
1495 [
1496 explicit(session(), Timeframe::Min30),
1497 explicit(visible(), Timeframe::Min30),
1498 ],
1499 400,
1500 2,
1501 ),
1502 (
1503 "explicit 15 minute session, 30 minute visible range",
1504 suggested(&[]),
1505 [
1506 explicit(session(), Timeframe::Min15),
1507 explicit(visible(), Timeframe::Min30),
1508 ],
1509 400,
1510 22,
1511 ),
1512 ] {
1513 let feed = Arc::new(RetryFeed::default());
1514 let mut chart = ChartBuilder::new()
1515 .with_feed(feed.clone())
1516 .with_symbol("MAIN")
1517 .with_timeframe(Timeframe::Hour1)
1518 .build();
1519 let id = studies.map(|study| chart.add_indicator(study))[1];
1520 let ctx = egui::Context::default();
1521 let mut befores = Vec::new();
1522 for frame in 0..150 {
1523 shapes(&ctx, &mut chart, frame as f64, Vec::new());
1524 for reply in feed.resolves.lock().unwrap().drain(..) {
1525 reply.send(Ok(instrument.clone()));
1526 }
1527 let pending: Vec<_> = feed.histories.lock().unwrap().drain(..).collect();
1528 for (request, reply) in pending {
1529 let minutes = request.timeframe.duration_ms() / 60_000;
1530 let before = request.before.unwrap_or(hour(400));
1531 if minutes == 60 {
1532 if request.before.is_some() && frame < 60 {
1534 feed.histories.lock().unwrap().push((request, reply));
1535 continue;
1536 }
1537 let end = hourly.partition_point(|bar| bar.time < before);
1538 let start = end.saturating_sub(main_page);
1539 let page = hourly[start..end].to_vec();
1540 reply.send(Ok(HistoryPage::new(page, start == 0)));
1541 continue;
1542 }
1543 if minutes == 30 {
1544 befores.push(request.before);
1545 }
1546 let page: Vec<_> = bars(&vec![1.0; 24_000 / minutes as usize], minutes)
1547 .into_iter()
1548 .filter(|bar| bar.time < before && bar.time >= before - Duration::hours(10))
1549 .collect();
1550 let end = page.first().is_none_or(|bar| bar.time == hour(0));
1551 reply.send(Ok(HistoryPage::new(page, end)));
1552 }
1553 }
1554 let held = chart
1555 .study_data
1556 .series
1557 .iter()
1558 .find(|series| series.request.timeframe == Some(Timeframe::Min30))
1559 .map_or(0, |series| {
1560 series
1561 .bars
1562 .iter()
1563 .filter(|bar| bar.time >= hour(305))
1564 .count()
1565 });
1566 let profile = chart.indicators().find(|(key, _)| *key == id).unwrap().1;
1567 let volume: f64 = profile
1568 .volume_profiles(305..400)
1569 .iter()
1570 .flat_map(|plot| plot.profile.volume())
1571 .sum();
1572 assert_eq!(
1573 (
1574 befores.iter().filter(|before| before.is_none()).count(),
1575 befores.last().copied().flatten(),
1576 held,
1577 volume,
1578 ),
1579 (1, Some(hour(12 * 24 + last)), 190, 190.0),
1580 "{name}"
1581 );
1582 }
1583 }
1584
1585 #[test]
1589 fn study_data_reaches_allocate_only_the_requests() {
1590 use crate::studies::{Indicator, VolProfileSession, VolProfileVisible};
1591 let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 30], 60)));
1592 chart.set_symbol("MAIN");
1593 chart.set_timeframe(Timeframe::Hour1);
1594 for mut profile in [
1595 Box::new(VolProfileSession::new(10)) as Box<dyn Indicator>,
1596 Box::new(VolProfileVisible::new(10)),
1597 ] {
1598 profile
1599 .set_input(
1600 "source_interval",
1601 InputValue::Timeframe(Some(Timeframe::Min30)),
1602 )
1603 .unwrap();
1604 chart.add_indicator(profile);
1605 }
1606 let ctx = egui::Context::default();
1607 for frame in 0..3 {
1608 shapes(&ctx, &mut chart, frame as f64, Vec::new());
1609 }
1610 let (requests, expected) = allocations(|| chart.study_data_requests());
1611 let (reaches, resolved) = allocations(|| chart.study_data_reaches());
1612 assert_eq!(reaches, requests);
1613 assert_eq!(
1614 resolved,
1615 expected
1616 .into_iter()
1617 .map(|request| (request, Some(DateTime::UNIX_EPOCH)))
1618 .collect::<Vec<_>>()
1619 );
1620 }
1621
1622 #[cfg(feature = "drawings")]
1624 #[test]
1625 fn drawing_profile_history_stops_at_the_profile_span() {
1626 use crate::drawings::{ChartPoint, Drawing, DrawingToolType};
1627 use crate::model::{ChartType, ProfileResolution};
1628 use crate::studies::Indicator;
1629 let feed = Arc::new(RetryFeed::default());
1630 let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 10], 60)));
1631 chart.set_symbol("MAIN");
1632 chart.set_timeframe(Timeframe::Hour1);
1633 chart.set_instrument(Some(Instrument::new("MAIN").with_profile_resolutions([
1634 ProfileResolution::new(Timeframe::Min30, Some(Duration::hours(12))),
1635 ])));
1636 let mut drawing = Drawing::new(1, DrawingToolType::FixedRangeVolumeProfile);
1637 drawing.chart_points = vec![ChartPoint::new(7.0, 1.0), ChartPoint::new(8.0, 1.0)];
1638 chart.drawings_mut().drawings.push(drawing);
1639 let request = StudyDataRequest {
1640 symbol: "MAIN".into(),
1641 timeframe: Some(Timeframe::Min30),
1642 metric: None,
1643 };
1644 assert_eq!(
1645 chart.study_data_requests().as_slice(),
1646 std::slice::from_ref(&request)
1647 );
1648 let ctx = egui::Context::default();
1649 let fine = bars(&[1.0; 20], 30);
1650 let hour = |h| Some(DateTime::UNIX_EPOCH + Duration::hours(h));
1651 for (anchor, expected) in [
1652 (7.0, vec![None, hour(9), hour(8)]),
1653 (5.0, vec![hour(7), hour(6)]),
1654 ] {
1655 chart.drawings_mut().drawings[0].chart_points[0].bar_idx = anchor;
1656 let mut befores = Vec::new();
1657 loop {
1658 ctx.run_ui(egui::RawInput::default(), |_| {})
1659 .drop_without_applying_deltas();
1660 chart.pump_study_data(&ctx, Some(feed.clone()));
1661 if let Some(reply) = feed.resolves.lock().unwrap().pop() {
1662 reply.send(Ok(Instrument::new("MAIN")));
1663 continue;
1664 }
1665 let Some((asked, reply)) = feed.histories.lock().unwrap().pop() else {
1666 break;
1667 };
1668 befores.push(asked.before);
1669 let end =
1670 fine.partition_point(|bar| asked.before.is_none_or(|before| bar.time < before));
1671 reply.send(Ok(HistoryPage::new(
1672 fine[end.saturating_sub(2)..end].to_vec(),
1673 end <= 2,
1674 )));
1675 }
1676 assert_eq!(befores, expected, "anchor {anchor}");
1677 }
1678 let mut study = crate::studies::VolProfileFixed::new(2);
1680 study
1681 .set_input(
1682 "source_interval",
1683 InputValue::Timeframe(Some(Timeframe::Min30)),
1684 )
1685 .unwrap();
1686 study.set_timeframe(Timeframe::Hour1);
1687 study.calculate(&bars(&[1.0; 10], 60));
1688 study.set_visible_range(0..10);
1689 let id = chart.add_indicator(Box::new(study));
1690 assert_eq!(chart.study_data_requests(), [request]);
1691 assert_eq!(chart.study_data_reaches()[0].1, hour(0));
1692 chart.remove_series(id);
1693 chart.drawings_mut().drawings[0].visible = false;
1694 assert!(chart.study_data_requests().is_empty());
1695 chart.drawings_mut().drawings[0].visible = true;
1696 chart.set_chart_type(ChartType::Renko);
1697 assert!(chart.study_data_requests().is_empty());
1698 }
1699}