Skip to main content

egui_charts/widget/
studies_data.rs

1//! Host-fed histories and subscriptions used by studies.
2
3use 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    /// Kept intact until success, including while newer live data arrives.
37    history: Option<HistoryRequest>,
38    /// The chart's currency or unit conversion, which `clear_feed` replaces the stream on.
39    conversion: Option<String>,
40    /// Where older bars continue after an empty page.
41    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    /// Histories required by feed-backed series, study inputs and profile drawings, with chart
201    /// defaults resolved. Hosts without a connected feed supply these through [`Self::set_study_data`]. A connected
202    /// feed receives resolve, history and subscription calls automatically on the next frame.
203    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    /// Replace a study history with host-owned historical or streaming bars. Bars are sorted,
224    /// deduplicated and validated like chart data. This takes ownership of this request from
225    /// the connected feed; the host supplies further replacements. Metadata sets its calendar.
226    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    /// The last error for a requested study history. Recoverable resolve and page failures retry
251    /// with exponential backoff, respecting rate limits; success clears the error. An unavailable
252    /// source leaves gaps.
253    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    /// [`Self::study_data_requests`], each with how far back the connected feed loads it: to
275    /// the chart's first bar, or to a later start that every study reading it states through
276    /// `Indicator::data_start`. `None` while the chart has no bars. Allocates only the requests,
277    /// since it runs every frame.
278    fn study_data_reaches(&self) -> Vec<(StudyDataRequest, Option<chrono::DateTime<chrono::Utc>>)> {
279        // Price-based chart types run their studies on elements; the source studies asked
280        // here are not given the view, so their histories load in full.
281        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                    // `None` sorts first: a study without a start loads it to the first bar.
293                    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    /// Counts the heap allocations of each thread, so a test can bound its own.
669    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    // SAFETY: every call forwards to `System` with the caller's arguments; the counter only observes.
679    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            // SAFETY: forwarded unchanged.
683            unsafe { std::alloc::System.alloc(layout) }
684        }
685
686        unsafe fn dealloc(&self, ptr: *mut u8, layout: std::alloc::Layout) {
687            // SAFETY: forwarded unchanged.
688            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            // SAFETY: forwarded unchanged.
699            unsafe { std::alloc::System.realloc(ptr, layout, new_size) }
700        }
701    }
702
703    /// The heap allocations `run` makes on this thread, not counting the drop of its result.
704    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                // A page from an earlier seek must not change the current request.
886                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    /// Metric streams keep exact publication times, refresh dependent studies, and use
1038    /// the normal subscription lifetime when the chart's instrument changes.
1039    #[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        // Hourly SMA(2) first closes at 02:00 with (10+14)/2=12. The newest value is (14+20)/2=17.
1162        assert_eq!(output(&chart), [None, None, None, Some(12.0), Some(12.0)]);
1163        // The forming 02:30 bar has not reached 03:00, so its scheduled close cannot
1164        // confirm the 02:00 hourly value while waiting for the interval close.
1165        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    /// Under replay, a study on another symbol or interval reads only the bars its history
1285    /// completed by the cursor. The chart's unfinished bar must not show the closing value of the
1286    /// other history's bar in the same slot, whichever way the cursor moves; leaving replay
1287    /// restores the full history.
1288    #[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        // The other symbol's hour from 01:00 is unfinished at 01:30, so it has no value there.
1338        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    /// A volume profile on a finer interval loads that history back to the first bar its
1379    /// profiles in view read, not to the chart's first bar, and loads more only when that bar
1380    /// moves earlier. Price-based chart types keep loading to the chart's first bar.
1381    #[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        // Half-hour bars from 00:00 to 09:30, answered two at a time from the newest.
1399        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    /// Guards #1667 and #1681: each full recompute of the profiles that read histories runs on
1442    /// its own frame, and a profile waiting for its turn keeps requesting the finer history it
1443    /// read and the start it read it from. So that history is loaded once from the newest bar,
1444    /// only back to the earliest profile start in view (the session open at 1970-01-13 00:00, or
1445    /// bar 305 at 17:00 for the visible range alone), and the recompute reads all 190 half-hour
1446    /// bars of volume 1 under bars 305 to 399. The feed answers ten hours a page.
1447    #[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                        // Older chart bars are held until the half-hour history has loaded.
1533                        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    /// Guards #1668: the pass that runs every frame to resolve the study histories also works
1586    /// out how far back each loads, allocating for the requests alone. Two profiles read one
1587    /// finer history here; the session profile reads the half-hour bars from 00:00.
1588    #[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    /// Drawings request one finer history and page only to their earliest visible anchor.
1623    #[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        // A study sharing the interval still controls its own earlier reach.
1679        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}