studies_data.rs

//! Host-fed histories and subscriptions used by studies.

use super::Chart;
use crate::data::inbox::Inbox;
use crate::data::merge::{Origin, merge};
use crate::data::{
    Datafeed, FeedError, HistoryPage, HistoryRequest, MetricPage, MetricRequest,
    MetricSubscription, Subscription, SubscriptionId,
};
use crate::model::{Bar, BarClock, Instrument, Timeframe};
use crate::studies::{StudyData, StudyDataRequest};
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};

#[derive(Default)]
pub(super) struct StudyDataCache {
    series: Vec<Series>,
}

struct Series {
    request: StudyDataRequest,
    bars: Vec<Bar>,
    instrument: Option<Instrument>,
    clock: BarClock,
    error: Option<FeedError>,
    stream: Option<Stream>,
    from_ticks: BTreeMap<chrono::DateTime<chrono::Utc>, Origin>,
}

struct Stream {
    feed: Arc<dyn Datafeed>,
    inbox: Arc<Inbox>,
    subscription: Option<SubscriptionId>,
    pending: bool,
    next_key: u64,
    /// Kept intact until success, including while newer live data arrives.
    history: Option<HistoryRequest>,
    /// The chart's currency or unit conversion, which `clear_feed` replaces the stream on.
    conversion: Option<String>,
    /// Where older bars continue after an empty page.
    cursor: Option<chrono::DateTime<chrono::Utc>>,
    retry_at: Option<f64>,
    failures: u32,
    end: bool,
    wake: Arc<Mutex<Option<egui::ViewportId>>>,
}

impl Stream {
    fn request(&mut self, source: &StudyDataRequest) {
        self.next_key += 1;
        self.pending = true;
        if let Some(history) = &self.history {
            if let Some(metric) = &source.metric {
                self.feed.metric_history(
                    MetricRequest {
                        symbol: history.symbol.clone(),
                        metric: metric.clone(),
                        before: history.before,
                        count: history.count,
                    },
                    self.inbox.reply(self.next_key),
                );
            } else {
                self.feed
                    .history(history.clone(), self.inbox.reply(self.next_key));
            }
        } else {
            super::feed::resolve(
                &*self.feed,
                &source.symbol,
                self.conversion.as_deref(),
                self.inbox.reply(self.next_key),
            );
        }
    }

    fn failed(&mut self, error: &FeedError, now: f64) {
        if error.is_recoverable() {
            let delay = super::feed::retry_delay(self.failures, error);
            self.retry_at = Some(now + delay.as_secs_f64());
            self.failures = self.failures.saturating_add(1);
        }
    }
}

impl Drop for Stream {
    fn drop(&mut self) {
        *self.wake.lock().unwrap_or_else(|error| error.into_inner()) = None;
        if let Some(id) = self.subscription {
            self.inbox.unsubscribe(id);
            self.feed.unsubscribe(id);
        }
    }
}

fn resolved(
    request: &StudyDataRequest,
    symbol: &str,
    timeframe: Option<Timeframe>,
) -> StudyDataRequest {
    StudyDataRequest {
        metric: request.metric.clone(),
        symbol: if request.symbol.is_empty() {
            symbol.to_owned()
        } else {
            request.symbol.clone()
        },
        timeframe: Some(request.timeframe.or(timeframe).unwrap_or_default()),
    }
}

fn clock(request: &StudyDataRequest, instrument: Option<&Instrument>, extended: bool) -> BarClock {
    if request.metric.is_some() {
        return BarClock::new(Timeframe::Ticks(std::num::NonZeroU64::MIN));
    }
    let timeframe = request.timeframe.unwrap_or_default();
    let mut clock = BarClock::new(timeframe);
    if let Some(instrument) = instrument {
        let zone = instrument
            .time_zone
            .clone()
            .unwrap_or_else(crate::model::TimezoneId::utc);
        if timeframe.duration_ms() < Timeframe::Day1.duration_ms() {
            clock = clock.with_timezone(zone.clone());
        }
        if let Some(session) = &instrument.session {
            clock = clock.with_session(zone, session.clone(), extended);
        }
    }
    clock
}

impl StudyDataCache {
    pub(super) fn snapshot(&self) -> Self {
        Self {
            series: self
                .series
                .iter()
                .map(|series| Series {
                    request: series.request.clone(),
                    bars: series.bars.clone(),
                    instrument: series.instrument.clone(),
                    clock: series.clock.clone(),
                    error: series.error.clone(),
                    stream: None,
                    from_ticks: series.from_ticks.clone(),
                })
                .collect(),
        }
    }
    pub(super) fn clear_feed(&mut self) {
        self.series.retain(|series| series.stream.is_none());
    }
    pub(super) fn pause(&mut self) {
        for series in &mut self.series {
            if let Some(stream) = &mut series.stream {
                stream.pending = false;
                stream.retry_at = None;
                stream.history = None;
                *stream
                    .wake
                    .lock()
                    .unwrap_or_else(|error| error.into_inner()) = None;
                if let Some(id) = stream.subscription.take() {
                    stream.inbox.unsubscribe(id);
                    stream.feed.unsubscribe(id);
                }
            }
        }
    }

    pub(super) fn data(
        &self,
        requests: Vec<StudyDataRequest>,
        symbol: &str,
        timeframe: Option<Timeframe>,
        replay: Option<chrono::DateTime<chrono::Utc>>,
    ) -> Vec<StudyData<'_>> {
        requests
            .into_iter()
            .filter_map(|request| {
                let key = resolved(&request, symbol, timeframe);
                let series = self.series.iter().find(|series| series.request == key)?;
                Some(StudyData {
                    request,
                    bars: match replay {
                        Some(time) => super::replay::completed(&series.bars, &series.clock, time),
                        None => &series.bars,
                    },
                    clock: series.clock.clone(),
                    instrument: series.instrument.as_ref(),
                    timeframe: key.timeframe.unwrap(),
                })
            })
            .collect()
    }
}

impl Chart {
    /// Histories required by feed-backed series, study inputs and profile drawings, with chart
    /// defaults resolved. Hosts without a connected feed supply these through [`Self::set_study_data`]. A connected
    /// feed receives resolve, history and subscription calls automatically on the next frame.
    pub fn study_data_requests(&self) -> Vec<StudyDataRequest> {
        let mut requests = self.fed_requests();
        for (_, study) in self.indicators() {
            for request in study.data_requests() {
                let request = resolved(&request, self.symbol(), self.timeframe());
                if !requests.contains(&request) {
                    requests.push(request);
                }
            }
        }
        #[cfg(feature = "drawings")]
        for (request, _) in self.profile_histories(&self.drawings) {
            let request = resolved(&request, self.symbol(), self.timeframe());
            if !requests.contains(&request) {
                requests.push(request);
            }
        }
        requests
    }

    /// Replace a study history with host-owned historical or streaming bars. Bars are sorted,
    /// deduplicated and validated like chart data. This takes ownership of this request from
    /// the connected feed; the host supplies further replacements. Metadata sets its calendar.
    pub fn set_study_data(
        &mut self,
        request: StudyDataRequest,
        bars: Vec<Bar>,
        instrument: Option<Instrument>,
    ) {
        let request = resolved(&request, self.symbol(), self.timeframe());
        let clock = clock(&request, instrument.as_ref(), self.extended_hours());
        let (bars, _) = super::series::clean(bars, &clock);
        self.study_data
            .series
            .retain(|series| series.request != request);
        self.study_data.series.push(Series {
            request,
            bars,
            instrument,
            clock,
            error: None,
            stream: None,
            from_ticks: BTreeMap::new(),
        });
        self.invalidate_study_data();
    }

    /// The last error for a requested study history. Recoverable resolve and page failures retry
    /// with exponential backoff, respecting rate limits; success clears the error. An unavailable
    /// source leaves gaps.
    pub fn study_data_error(&self, request: &StudyDataRequest) -> Option<&FeedError> {
        let request = resolved(request, self.symbol(), self.timeframe());
        self.study_data
            .series
            .iter()
            .find(|series| series.request == request)?
            .error
            .as_ref()
    }

    pub(super) fn invalidate_study_data(&mut self) {
        for (at, study) in self.studies.indicators().iter().enumerate() {
            if !study.data_requests().is_empty() {
                self.pending_from[at] = Some(0);
                self.next_refresh[at] = f64::NEG_INFINITY;
            }
        }
        self.element_studies.dirty = true;
        self.fill_fed_series();
    }

    /// [`Self::study_data_requests`], each with how far back the connected feed loads it: to
    /// the chart's first bar, or to a later start that every study reading it states through
    /// `Indicator::data_start`. `None` while the chart has no bars. Allocates only the requests,
    /// since it runs every frame.
    fn study_data_reaches(&self) -> Vec<(StudyDataRequest, Option<chrono::DateTime<chrono::Utc>>)> {
        // Price-based chart types run their studies on elements; the source studies asked
        // here are not given the view, so their histories load in full.
        let full = self.chart_type.is_time_independent();
        let mut reaches: Vec<(StudyDataRequest, Option<_>)> = self
            .fed_requests()
            .into_iter()
            .map(|request| (request, None))
            .collect();
        for (_, study) in self.indicators() {
            for declared in study.data_requests() {
                let start = study.data_start(&declared).filter(|_| !full);
                let request = resolved(&declared, self.symbol(), self.timeframe());
                match reaches.iter_mut().find(|(known, _)| *known == request) {
                    // `None` sorts first: a study without a start loads it to the first bar.
                    Some((_, reach)) => *reach = (*reach).min(start),
                    None => reaches.push((request, start)),
                }
            }
        }
        #[cfg(feature = "drawings")]
        for (declared, start) in self.profile_histories(&self.drawings) {
            let request = resolved(&declared, self.symbol(), self.timeframe());
            match reaches.iter_mut().find(|(known, _)| *known == request) {
                Some((_, reach)) => *reach = (*reach).min(Some(start)),
                None => reaches.push((request, Some(start))),
            }
        }
        let first = self.state.data().bars.first().map(|bar| bar.time);
        for (_, reach) in &mut reaches {
            *reach = first.map(|first| reach.map_or(first, |start| start.max(first)));
        }
        reaches
    }

    pub(super) fn pump_study_data(&mut self, ctx: &egui::Context, feed: Option<Arc<dyn Datafeed>>) {
        let requests = self.study_data_reaches();
        let previous = self.study_data.series.len();
        self.study_data.series.retain(|series| {
            requests
                .iter()
                .any(|(request, _)| *request == series.request)
                && series.stream.as_ref().is_none_or(|stream| {
                    feed.as_ref()
                        .is_some_and(|feed| Arc::ptr_eq(feed, &stream.feed))
                })
        });
        let removed = self.study_data.series.len() != previous;
        if let Some(feed) = &feed {
            for (request, _) in &requests {
                if request.symbol.is_empty() {
                    continue;
                }
                if self
                    .study_data
                    .series
                    .iter()
                    .any(|series| &series.request == request)
                {
                    continue;
                }
                let inbox = Arc::new(Inbox::default());
                let wake = Arc::new(Mutex::new(Some(ctx.viewport_id())));
                let repaint = Arc::clone(&wake);
                let ctx = ctx.clone();
                inbox.set_waker(move || {
                    let viewport = *repaint.lock().unwrap_or_else(|error| error.into_inner());
                    if let Some(viewport) = viewport {
                        ctx.request_repaint_of(viewport);
                    }
                });
                super::feed::resolve(
                    &**feed,
                    &request.symbol,
                    self.conversion.as_deref(),
                    inbox.reply(0),
                );
                self.study_data.series.push(Series {
                    request: request.clone(),
                    bars: Vec::new(),
                    instrument: None,
                    clock: clock(request, None, self.extended_hours()),
                    error: None,
                    from_ticks: BTreeMap::new(),
                    stream: Some(Stream {
                        feed: Arc::clone(feed),
                        inbox,
                        subscription: None,
                        pending: true,
                        next_key: 0,
                        history: None,
                        conversion: self.conversion.clone(),
                        cursor: None,
                        retry_at: None,
                        failures: 0,
                        end: false,
                        wake,
                    }),
                });
            }
        }
        let extended = self.extended_hours();
        let adjusted = self.dividend_adjusted();
        let now = ctx.input(|input| input.time);
        let count = self.state.data().bars.len().max(300);
        let mut changed = removed || self.study_data.series.len() != previous;
        for series in &mut self.study_data.series {
            let Some(stream) = &mut series.stream else {
                let clock = clock(&series.request, series.instrument.as_ref(), extended);
                if clock != series.clock {
                    series.clock = clock;
                    changed = true;
                }
                continue;
            };
            *stream
                .wake
                .lock()
                .unwrap_or_else(|error| error.into_inner()) = Some(ctx.viewport_id());
            let drained = stream.inbox.drain();
            for (key, answer) in drained.answers {
                if !stream.pending || key != stream.next_key {
                    continue;
                }
                stream.pending = false;
                if stream.history.is_none() {
                    let result = answer
                        .downcast::<Result<Instrument, FeedError>>()
                        .map(|result| *result)
                        .unwrap_or(Err(FeedError::NotSupported));
                    match result {
                        Ok(instrument) => {
                            let tf = series.request.timeframe.unwrap();
                            if series.request.metric.is_none()
                                && instrument
                                    .supported_timeframes
                                    .as_ref()
                                    .is_some_and(|intervals| !intervals.contains(&tf))
                            {
                                series.error = Some(FeedError::NotSupported);
                                continue;
                            }
                            series.clock = clock(&series.request, Some(&instrument), extended);
                            series.instrument = Some(instrument);
                            series.error = None;
                            stream.failures = 0;
                        }
                        Err(error) => {
                            stream.failed(&error, now);
                            series.error = Some(error);
                        }
                    }
                } else {
                    let result = if series.request.metric.is_some() {
                        answer
                            .downcast::<Result<MetricPage, FeedError>>()
                            .map(|result| {
                                result.map(|page| {
                                    HistoryPage::new(
                                        page.points.into_iter().map(Bar::from).collect(),
                                        page.end_of_history,
                                    )
                                })
                            })
                            .unwrap_or(Err(FeedError::NotSupported))
                    } else {
                        answer
                            .downcast::<Result<HistoryPage, FeedError>>()
                            .map(|result| *result)
                            .unwrap_or(Err(FeedError::NotSupported))
                    };
                    match result {
                        Ok(mut page) => {
                            let previous = stream.history.take().and_then(|request| request.before);
                            page.bars.retain(|bar| {
                                previous
                                    .is_none_or(|before| series.clock.bar_open(bar.time) < before)
                            });
                            series.error = None;
                            stream.failures = 0;
                            let empty = page.bars.is_empty();
                            let continuation = page.continue_before.filter(|_| empty);
                            stream.cursor = continuation;
                            stream.end =
                                page.end_of_history || (empty && page.continue_before.is_none());
                            if continuation.is_some_and(|at| previous.is_some_and(|b| at >= b)) {
                                stream.cursor = None;
                                stream.end = true;
                                series.error = Some(FeedError::InvalidData(
                                    "history continuation does not lie before the request".into(),
                                ));
                            }
                            let incoming = page
                                .bars
                                .into_iter()
                                .map(|bar| (bar, Origin::History))
                                .collect();
                            merge(
                                &mut series.bars,
                                &mut series.from_ticks,
                                &series.clock,
                                incoming,
                            );
                            if previous.is_some()
                                && continuation.is_none()
                                && series.bars.first().map(|bar| bar.time) >= previous
                            {
                                stream.end = true;
                            }
                            changed = true;
                        }
                        Err(error) => {
                            stream.failed(&error, now);
                            series.error = Some(error);
                        }
                    }
                }
            }
            for (_, updates) in drained.updates {
                if updates.reset {
                    series.bars.clear();
                    series.from_ticks.clear();
                    series.error = None;
                    stream.pending = false;
                    stream.history = None;
                    stream.retry_at = None;
                    stream.failures = 0;
                    stream.end = false;
                    changed = true;
                }
                if !updates.bars.is_empty() {
                    merge(
                        &mut series.bars,
                        &mut series.from_ticks,
                        &series.clock,
                        updates.bars.into_values().collect(),
                    );
                    changed = true;
                }
                if updates.error.is_some() {
                    series.error = updates.error;
                }
            }
            if stream.subscription.is_none()
                && series.instrument.is_some()
                && series.error.is_none()
            {
                let tf = series.request.timeframe.unwrap();
                let sink = stream.inbox.subscribe(
                    series.request.symbol.clone(),
                    series.clock.clone(),
                    super::feed::TRADE_CAPACITY,
                    series.instrument.as_ref().and_then(|i| i.tick_size),
                );
                stream.subscription = Some(sink.id);
                if let Some(metric) = &series.request.metric {
                    stream.feed.subscribe_metric(MetricSubscription {
                        id: sink.id,
                        symbol: series.request.symbol.clone(),
                        metric: metric.clone(),
                        sink,
                    });
                } else {
                    stream.feed.subscribe(Subscription {
                        id: sink.id,
                        symbol: series.request.symbol.clone(),
                        timeframe: tf,
                        clock: series.clock.clone(),
                        extended_hours: extended,
                        conversion: stream.conversion.clone(),
                        sink,
                    });
                }
            }
            if let Some(deadline) = stream.retry_at {
                if now >= deadline {
                    stream.retry_at = None;
                    stream.request(&series.request);
                } else {
                    ctx.request_repaint_after(
                        std::time::Duration::try_from_secs_f64(deadline - now)
                            .unwrap_or(std::time::Duration::MAX),
                    );
                }
            }
            let reach = requests
                .iter()
                .find(|(request, _)| *request == series.request)
                .and_then(|(_, reach)| *reach);
            if !stream.pending
                && stream.retry_at.is_none()
                && !stream.end
                && series.error.is_none()
                && series.instrument.is_some()
                && (series.bars.is_empty()
                    || reach
                        .zip(series.bars.first())
                        .is_some_and(|(reach, bar)| reach < bar.time))
            {
                stream.history = Some(HistoryRequest {
                    symbol: series.request.symbol.clone(),
                    timeframe: series.request.timeframe.unwrap(),
                    clock: series.clock.clone(),
                    before: series
                        .bars
                        .first()
                        .map(|bar| bar.time)
                        .into_iter()
                        .chain(stream.cursor)
                        .min(),
                    since: None,
                    count,
                    extended_hours: extended,
                    adjusted,
                    conversion: stream.conversion.clone(),
                });
                stream.request(&series.request);
            }
        }
        if changed {
            self.invalidate_study_data();
        }
    }
}

#[cfg(all(test, feature = "studies"))]
mod tests {
    use super::*;
    use crate::data::{Reply, Sink};
    use crate::studies::{InputValue, SMA};
    use crate::widget::{ChartBuilder, test_frames::shapes};
    use chrono::{DateTime, Duration};

    #[derive(Default)]
    struct Feed {
        sinks: Mutex<Vec<(String, Sink)>>,
        correction: Mutex<f64>,
        metrics: Mutex<Vec<MetricSubscription>>,
    }
    impl Datafeed for Feed {
        fn history(&self, request: HistoryRequest, reply: Reply<HistoryPage>) {
            let (prices, minutes) = match request.symbol.as_str() {
                "REMOTE" => (vec![10.0, 14.0, 20.0], 60),
                "OTHER" => (
                    vec![100.0, 200.0, 300.0 + *self.correction.lock().unwrap()],
                    60,
                ),
                _ => (vec![1.0; 5], 30),
            };
            reply.send(Ok(HistoryPage::new(bars(&prices, minutes), true)));
        }
        fn metric_history(&self, _: MetricRequest, reply: Reply<MetricPage>) {
            reply.send(Ok(MetricPage {
                points: [(15, 10.0), (75, 20.0)]
                    .into_iter()
                    .map(|(minute, value)| crate::data::MetricPoint {
                        time: DateTime::UNIX_EPOCH + Duration::minutes(minute),
                        value,
                    })
                    .collect(),
                end_of_history: true,
            }));
        }
        fn subscribe_metric(&self, subscription: MetricSubscription) {
            self.metrics.lock().unwrap().push(subscription);
        }
        fn subscribe(&self, subscription: Subscription) {
            self.sinks
                .lock()
                .unwrap()
                .push((subscription.symbol, subscription.sink));
        }
    }
    fn bars(prices: &[f64], minutes: i64) -> Vec<Bar> {
        prices
            .iter()
            .enumerate()
            .map(|(at, &price)| {
                Bar::new(
                    DateTime::UNIX_EPOCH + Duration::minutes(at as i64 * minutes),
                    price,
                    price,
                    price,
                    price,
                    1.0,
                )
            })
            .collect()
    }

    /// Counts the heap allocations of each thread, so a test can bound its own.
    struct Counting;

    thread_local! {
        static ALLOCATIONS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    }

    #[global_allocator]
    static ALLOCATOR: Counting = Counting;

    // SAFETY: every call forwards to `System` with the caller's arguments; the counter only observes.
    unsafe impl std::alloc::GlobalAlloc for Counting {
        unsafe fn alloc(&self, layout: std::alloc::Layout) -> *mut u8 {
            let _ = ALLOCATIONS.try_with(|count| count.set(count.get() + 1));
            // SAFETY: forwarded unchanged.
            unsafe { std::alloc::System.alloc(layout) }
        }

        unsafe fn dealloc(&self, ptr: *mut u8, layout: std::alloc::Layout) {
            // SAFETY: forwarded unchanged.
            unsafe { std::alloc::System.dealloc(ptr, layout) }
        }

        unsafe fn realloc(
            &self,
            ptr: *mut u8,
            layout: std::alloc::Layout,
            new_size: usize,
        ) -> *mut u8 {
            let _ = ALLOCATIONS.try_with(|count| count.set(count.get() + 1));
            // SAFETY: forwarded unchanged.
            unsafe { std::alloc::System.realloc(ptr, layout, new_size) }
        }
    }

    /// The heap allocations `run` makes on this thread, not counting the drop of its result.
    fn allocations<T>(run: impl FnOnce() -> T) -> (usize, T) {
        let before = ALLOCATIONS.with(std::cell::Cell::get);
        let result = run();
        (ALLOCATIONS.with(std::cell::Cell::get) - before, result)
    }

    #[derive(Default)]
    struct RetryFeed {
        resolves: Mutex<Vec<crate::data::Reply<Instrument>>>,
        histories: Mutex<Vec<(HistoryRequest, crate::data::Reply<HistoryPage>)>>,
        metrics: Mutex<Vec<(MetricRequest, crate::data::Reply<MetricPage>)>>,
        sinks: Mutex<Vec<Sink>>,
    }
    impl Datafeed for RetryFeed {
        fn resolve(&self, _: &str, reply: Reply<Instrument>) {
            self.resolves.lock().unwrap().push(reply);
        }
        fn history(&self, request: HistoryRequest, reply: Reply<HistoryPage>) {
            self.histories.lock().unwrap().push((request, reply));
        }
        fn metric_history(&self, request: MetricRequest, reply: Reply<MetricPage>) {
            self.metrics.lock().unwrap().push((request, reply));
        }
        fn subscribe(&self, subscription: Subscription) {
            self.sinks.lock().unwrap().push(subscription.sink);
        }
        fn subscribe_metric(&self, subscription: MetricSubscription) {
            self.sinks.lock().unwrap().push(subscription.sink);
        }
    }
    impl RetryFeed {
        fn pending(&self, metric: bool) -> usize {
            if metric {
                self.metrics.lock().unwrap().len()
            } else {
                self.histories.lock().unwrap().len()
            }
        }
        fn answer(
            &self,
            metric: bool,
            result: Result<HistoryPage, FeedError>,
        ) -> (Option<DateTime<chrono::Utc>>, usize) {
            if metric {
                let (request, reply) = self.metrics.lock().unwrap().remove(0);
                reply.send(result.map(|page| {
                    MetricPage {
                        points: page
                            .bars
                            .into_iter()
                            .map(|bar| crate::data::MetricPoint {
                                time: bar.time,
                                value: bar.close,
                            })
                            .collect(),
                        end_of_history: page.end_of_history,
                    }
                }));
                (request.before, request.count)
            } else {
                let (request, reply) = self.histories.lock().unwrap().remove(0);
                reply.send(result);
                (request.before, request.count)
            }
        }
    }

    #[test]
    fn study_sources_retry_original_requests_and_ignore_superseded_replies() {
        for metric in [false, true] {
            for (error, delay) in [
                (FeedError::Network("offline".into()), 1.0),
                (
                    FeedError::RateLimited {
                        retry_after: Some(std::time::Duration::from_secs(7)),
                    },
                    7.0,
                ),
                (FeedError::Unanswered, 1.0),
                (FeedError::NotSupported, 0.0),
            ] {
                let feed = Arc::new(RetryFeed::default());
                let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 5], 30)));
                chart.set_symbol("REMOTE");
                if metric {
                    chart.add_indicator(Box::new(crate::studies::MetricStudy::new(
                        crate::data::MetricSeries {
                            id: "oi".into(),
                            name: "Open interest".into(),
                            unit: "contracts".into(),
                            format: crate::scales::StudyFormat::Volume,
                        },
                    )));
                } else {
                    let id = chart.add_indicator(Box::new(SMA::new(1)));
                    chart
                        .indicator_mut(id)
                        .unwrap()
                        .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
                        .unwrap();
                }
                let ctx = egui::Context::default();
                let pump = |chart: &mut Chart, time| {
                    ctx.run_ui(
                        egui::RawInput {
                            time: Some(time),
                            ..Default::default()
                        },
                        |_| {},
                    )
                    .drop_without_applying_deltas();
                    chart.pump_study_data(&ctx, Some(feed.clone()));
                };
                pump(&mut chart, 0.0);
                feed.resolves
                    .lock()
                    .unwrap()
                    .pop()
                    .unwrap()
                    .send(Err(error.clone()));
                pump(&mut chart, 0.25);
                let request = chart.study_data_requests().pop().unwrap();
                assert_eq!(chart.study_data_error(&request), Some(&error));
                pump(&mut chart, 0.25 + delay - 0.01);
                assert!(feed.resolves.lock().unwrap().is_empty());
                pump(&mut chart, 0.25 + delay);
                if !error.is_recoverable() {
                    pump(&mut chart, 120.0);
                    assert!(feed.resolves.lock().unwrap().is_empty());
                    continue;
                }
                assert_eq!(feed.resolves.lock().unwrap().len(), 1, "{error:?}");
                feed.resolves
                    .lock()
                    .unwrap()
                    .pop()
                    .unwrap()
                    .send(Ok(Instrument::new("REMOTE")));
                let now = 0.5 + delay;
                pump(&mut chart, now);
                assert_eq!(chart.study_data_error(&request), None);
                let mut newest = bars(&[20.0], 60);
                newest[0].time += Duration::hours(2);
                assert_eq!(
                    feed.answer(metric, Ok(HistoryPage::new(newest, false))).0,
                    None
                );
                pump(&mut chart, now + 0.25);
                let failed = feed.answer(metric, Err(error.clone()));
                pump(&mut chart, now + 0.5);
                feed.sinks.lock().unwrap()[0].push_bar(Bar::new(
                    DateTime::UNIX_EPOCH + Duration::hours(1),
                    55.0,
                    55.0,
                    55.0,
                    55.0,
                    1.0,
                ));
                pump(&mut chart, now + 0.5 + delay - 0.01);
                assert_eq!(feed.pending(metric), 0);
                pump(&mut chart, now + 0.5 + delay);
                assert_eq!(feed.pending(metric), 1);
                let retried = feed.answer(
                    metric,
                    Ok(HistoryPage::new(bars(&[12.0, 55.0, 99.0, 100.0], 60), true)),
                );
                assert_eq!(
                    retried, failed,
                    "retry must keep its boundary and count after a live prepend"
                );
                assert_eq!(retried.0, Some(DateTime::UNIX_EPOCH + Duration::hours(2)));
                pump(&mut chart, now + 0.75 + delay);
                assert_eq!(chart.study_data_error(&request), None);
                assert_eq!(
                    chart.study_data.series[0]
                        .bars
                        .iter()
                        .map(|bar| bar.close)
                        .collect::<Vec<_>>(),
                    [12.0, 55.0, 20.0]
                );
                // A page from an earlier seek must not change the current request.
                for time in [now + 10.0, now + 11.0] {
                    feed.sinks.lock().unwrap()[0].reset_replay(DateTime::UNIX_EPOCH);
                    pump(&mut chart, time);
                }
                feed.answer(metric, Err(FeedError::Network("old page".into())));
                pump(&mut chart, now + 12.0);
                assert_eq!(chart.study_data_error(&request), None);
                feed.answer(metric, Ok(HistoryPage::new(bars(&[99.0], 60), true)));
                pump(&mut chart, now + 13.0);
                assert_eq!(chart.study_data_error(&request), None);
                assert_eq!(chart.study_data.series[0].bars[0].close, 99.0);
                assert_eq!(feed.pending(metric), 0);
            }
        }
    }

    #[test]
    fn study_retries_back_off_and_cancel_with_their_source() {
        for action in ["pause", "host data", "remove", "replace"] {
            let feed = Arc::new(RetryFeed::default());
            let mut chart = Chart::new(crate::model::BarData::new());
            let id = chart.add_indicator(Box::new(SMA::new(1)));
            chart
                .indicator_mut(id)
                .unwrap()
                .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
                .unwrap();
            let ctx = egui::Context::default();
            let pump = |chart: &mut Chart, time| {
                ctx.run_ui(
                    egui::RawInput {
                        time: Some(time),
                        ..Default::default()
                    },
                    |_| {},
                )
                .drop_without_applying_deltas();
                chart.pump_study_data(&ctx, Some(feed.clone()));
            };
            pump(&mut chart, 0.0);
            feed.resolves
                .lock()
                .unwrap()
                .pop()
                .unwrap()
                .send(Err(FeedError::Network("offline".into())));
            pump(&mut chart, 0.1);
            pump(&mut chart, 1.1);
            feed.resolves
                .lock()
                .unwrap()
                .pop()
                .unwrap()
                .send(Err(FeedError::Network("still offline".into())));
            pump(&mut chart, 1.2);
            pump(&mut chart, 2.2);
            assert!(
                feed.resolves.lock().unwrap().is_empty(),
                "second failure waits two seconds"
            );
            match action {
                "pause" => {
                    chart.set_feed_paused(true);
                }
                "host data" => {
                    let request = chart.study_data_requests().pop().unwrap();
                    chart.set_study_data(request, bars(&[99.0], 60), None);
                }
                "remove" => {
                    chart.remove_series(id);
                }
                "replace" => {
                    chart
                        .indicator_mut(id)
                        .unwrap()
                        .set_input("calculation_symbol", InputValue::Symbol("OTHER".into()))
                        .unwrap();
                }
                _ => unreachable!(),
            }
            pump(&mut chart, 60.0);
            assert_eq!(
                feed.resolves.lock().unwrap().len(),
                usize::from(action == "replace"),
                "{action}"
            );
        }
    }

    #[test]
    fn replay_reset_discards_future_study_bars_and_tick_origins() {
        for replacement in [false, true] {
            let feed = Arc::new(Feed::default());
            let mut chart = ChartBuilder::new()
                .with_feed(feed.clone())
                .with_symbol("MAIN")
                .with_timeframe(Timeframe::Hour1)
                .build();
            let id = chart.add_indicator(Box::new(SMA::new(1)));
            chart
                .indicator_mut(id)
                .unwrap()
                .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
                .unwrap();
            let ctx = egui::Context::default();
            for frame in 0..8 {
                shapes(&ctx, &mut chart, frame as f64, Vec::new());
            }
            let sink = feed
                .sinks
                .lock()
                .unwrap()
                .iter()
                .find(|(symbol, _)| symbol == "REMOTE")
                .unwrap()
                .1
                .clone();
            sink.set_server_time(DateTime::UNIX_EPOCH + Duration::hours(4));
            sink.push_tick(crate::model::Tick::new(
                DateTime::UNIX_EPOCH + Duration::hours(3),
                30.0,
                1.0,
                None,
            ));
            chart.pump_study_data(&ctx, Some(feed.clone()));
            assert!(!chart.study_data.series[0].from_ticks.is_empty());
            sink.reset_replay(DateTime::UNIX_EPOCH);
            if replacement {
                sink.push_tick(crate::model::Tick::new(
                    DateTime::UNIX_EPOCH,
                    99.0,
                    1.0,
                    None,
                ));
            }
            chart.pump_study_data(&ctx, Some(feed.clone()));
            let series = &chart.study_data.series[0];
            assert_eq!(series.bars.len(), usize::from(replacement));
            assert!(
                series
                    .from_ticks
                    .keys()
                    .all(|time| *time == DateTime::UNIX_EPOCH)
            );
            if replacement {
                assert_eq!(series.bars[0].close, 99.0);
            }
            assert_eq!(chart.pending_from[0], Some(0));
        }
    }

    /// Metric streams keep exact publication times, refresh dependent studies, and use
    /// the normal subscription lifetime when the chart's instrument changes.
    #[test]
    fn metric_history_and_live_points_refresh_the_pane_and_its_consumers() {
        use crate::data::{MetricPoint, MetricSeries};
        use crate::scales::StudyFormat;
        use crate::studies::{MetricStudy, StudySource};
        let feed = Arc::new(Feed::default());
        let mut chart = ChartBuilder::new()
            .with_feed(feed.clone())
            .with_symbol("MAIN")
            .with_timeframe(Timeframe::Min30)
            .build();
        let metric = chart.add_indicator(Box::new(MetricStudy::new(MetricSeries {
            id: "oi".into(),
            name: "Open interest".into(),
            unit: "contracts".into(),
            format: StudyFormat::Volume,
        })));
        let average = chart.add_indicator(Box::new(SMA::new(2)));
        chart
            .set_indicator_source(
                average,
                StudySource::Study {
                    id: metric.0,
                    plot: 0,
                },
            )
            .unwrap();
        let ctx = egui::Context::default();
        let settle = |chart: &mut Chart| {
            for frame in 0..12 {
                shapes(&ctx, chart, frame as f64, Vec::new());
            }
        };
        let output = |chart: &Chart, id| {
            let study = chart.indicators().find(|(key, _)| *key == id).unwrap().1;
            (0..5)
                .map(|at| study.values().value(0, at))
                .collect::<Vec<_>>()
        };
        settle(&mut chart);
        assert_eq!(
            output(&chart, metric),
            [None, Some(10.0), Some(10.0), Some(20.0), Some(20.0)]
        );
        assert_eq!(
            output(&chart, average),
            [None, None, Some(10.0), Some(15.0), Some(20.0)]
        );
        let sink = feed.metrics.lock().unwrap()[0].sink.clone();
        for (minutes, value) in [(15, 14.0), (110, 30.0)] {
            sink.push_metric(MetricPoint {
                time: DateTime::UNIX_EPOCH + Duration::minutes(minutes),
                value,
            });
        }
        for frame in 0..60 {
            let mut last = bars(&[1.0; 5], 30).pop().unwrap();
            last.close = 1.0 + frame as f64 * 0.001;
            last.high = last.close;
            chart.update(super::super::SeriesId::MAIN, last);
            shapes(&ctx, &mut chart, 12.0 + frame as f64 / 60.0, Vec::new());
        }
        assert_eq!(
            output(&chart, metric),
            [None, Some(14.0), Some(14.0), Some(20.0), Some(30.0)]
        );
        assert_eq!(
            output(&chart, average),
            [None, None, Some(14.0), Some(17.0), Some(25.0)]
        );
        chart.set_symbol("OTHER");
        settle(&mut chart);
        assert!(!sink.is_active());
        drop(chart);
        assert!(
            feed.metrics
                .lock()
                .unwrap()
                .iter()
                .all(|sub| !sub.sink.is_active())
        );
    }

    #[test]
    fn requested_intervals_align_confirmed_and_live_values_and_replace_subscriptions() {
        let feed = Arc::new(Feed::default());
        let mut chart = ChartBuilder::new()
            .with_feed(feed.clone())
            .with_symbol("MAIN")
            .with_timeframe(Timeframe::Min30)
            .build();
        let id = chart.add_indicator(Box::new(SMA::new(2)));
        let study = chart.indicator_mut(id).unwrap();
        study
            .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
            .unwrap();
        study
            .set_input(
                "calculation_timeframe",
                InputValue::Timeframe(Some(Timeframe::Hour1)),
            )
            .unwrap();
        let ctx = egui::Context::default();
        let settle = |chart: &mut Chart| {
            for frame in 0..10 {
                shapes(&ctx, chart, frame as f64, Vec::new());
            }
        };
        settle(&mut chart);
        let output = |chart: &Chart| {
            (0..5)
                .map(|at| {
                    chart
                        .indicators()
                        .find(|(candidate, _)| *candidate == id)
                        .unwrap()
                        .1
                        .values()
                        .value(0, at)
                })
                .collect::<Vec<_>>()
        };
        // Hourly SMA(2) first closes at 02:00 with (10+14)/2=12. The newest value is (14+20)/2=17.
        assert_eq!(output(&chart), [None, None, None, Some(12.0), Some(12.0)]);
        // The forming 02:30 bar has not reached 03:00, so its scheduled close cannot
        // confirm the 02:00 hourly value while waiting for the interval close.
        chart.set_data(super::super::SeriesId::MAIN, bars(&[1.0; 6], 30));
        settle(&mut chart);
        assert_eq!(
            chart
                .indicators()
                .find(|(candidate, _)| *candidate == id)
                .unwrap()
                .1
                .values()
                .value(0, 5),
            Some(12.0)
        );
        chart.set_data(super::super::SeriesId::MAIN, bars(&[1.0; 5], 30));
        settle(&mut chart);
        chart
            .indicator_mut(id)
            .unwrap()
            .set_input("wait_for_close", InputValue::Toggle(false))
            .unwrap();
        settle(&mut chart);
        assert_eq!(output(&chart), [None, None, None, Some(12.0), Some(17.0)]);
        let sink = feed
            .sinks
            .lock()
            .unwrap()
            .iter()
            .find(|(symbol, _)| symbol == "REMOTE")
            .unwrap()
            .1
            .clone();
        sink.push_bar(bars(&[10.0, 14.0, 24.0], 60).pop().unwrap());
        settle(&mut chart);
        assert_eq!(output(&chart)[4], Some(19.0));
        chart
            .indicator_mut(id)
            .unwrap()
            .set_input("calculation_symbol", InputValue::Symbol("OTHER".into()))
            .unwrap();
        settle(&mut chart);
        assert!(!sink.is_active());
        assert_eq!(output(&chart)[4], Some(250.0));
        let other_sink = feed
            .sinks
            .lock()
            .unwrap()
            .iter()
            .rev()
            .find(|(symbol, _)| symbol == "OTHER")
            .unwrap()
            .1
            .clone();
        chart.set_extended_hours(!chart.extended_hours());
        settle(&mut chart);
        assert!(!other_sink.is_active());
        assert_eq!(output(&chart)[4], Some(250.0));
        chart.set_feed_paused(true);
        *feed.correction.lock().unwrap() = 100.0;
        chart.set_feed_paused(false);
        settle(&mut chart);
        assert_eq!(
            output(&chart)[4],
            Some(300.0),
            "resuming reloads missed bars"
        );
        let request = chart.study_data_requests().pop().unwrap();
        chart.set_study_data(request, bars(&[2.0, 4.0, 8.0], 60), None);
        settle(&mut chart);
        assert_eq!(output(&chart)[4], Some(6.0));
        chart
            .indicator_mut(id)
            .unwrap()
            .set_input("wait_for_close", InputValue::Toggle(true))
            .unwrap();
        for (name, minutes, stamp, missing, expected) in [
            (
                "stamped inside hourly slots",
                60,
                10,
                None,
                [None, None, None, Some(3.0), Some(3.0)],
            ),
            ("missing hourly slot", 60, 0, Some(1), [None; 5]),
            (
                "missing same-interval slot",
                30,
                0,
                Some(1),
                [None, None, Some(5.0), None, None],
            ),
        ] {
            let timeframe = if minutes == 60 {
                Timeframe::Hour1
            } else {
                Timeframe::Min30
            };
            chart
                .indicator_mut(id)
                .unwrap()
                .set_input(
                    "calculation_timeframe",
                    InputValue::Timeframe(Some(timeframe)),
                )
                .unwrap();
            let source = bars(&[2.0, 4.0, 8.0], minutes)
                .into_iter()
                .enumerate()
                .filter(|(at, _)| Some(*at) != missing)
                .map(|(_, mut bar)| {
                    bar.time += Duration::minutes(stamp);
                    bar
                })
                .collect();
            let request = chart.study_data_requests().pop().unwrap();
            chart.set_study_data(request, source, None);
            settle(&mut chart);
            assert_eq!(output(&chart), expected, "{name}");
        }
    }

    /// Under replay, a study on another symbol or interval reads only the bars its history
    /// completed by the cursor. The chart's unfinished bar must not show the closing value of the
    /// other history's bar in the same slot, whichever way the cursor moves; leaving replay
    /// restores the full history.
    #[test]
    fn replay_hides_the_future_of_other_histories() {
        let hourly = bars(&[1.0; 3], 60);
        let mut chart = Chart::new(crate::model::BarData::from_bars(hourly));
        chart.set_symbol("LOCAL");
        chart.set_timeframe(Timeframe::Hour1);
        let id = chart.add_indicator(Box::new(SMA::new(1)));
        chart
            .indicator_mut(id)
            .unwrap()
            .set_input("calculation_symbol", InputValue::Symbol("REMOTE".into()))
            .unwrap();
        let request = chart.study_data_requests().pop().unwrap();
        chart.set_study_data(request, bars(&[10.0, 14.0, 24.0], 60), None);
        let ctx = egui::Context::default();
        let mut frame = 0.0;
        let mut values = |chart: &mut Chart| {
            for _ in 0..4 {
                frame += 1.0;
                shapes(&ctx, chart, frame, Vec::new());
            }
            let study = chart.indicators().next().unwrap().1;
            let shown: Vec<_> = (0..study.values().len())
                .map(|at| study.values().value(0, at))
                .collect();
            #[cfg(feature = "csv")]
            {
                let exported = chart.export_csv().unwrap();
                let mut reader = csv::Reader::from_reader(exported.as_bytes());
                let column = reader
                    .headers()
                    .unwrap()
                    .iter()
                    .position(|name| name == "SMA: SMA(1)")
                    .unwrap();
                let exported: Vec<Option<f64>> = reader
                    .records()
                    .map(|row| row.unwrap()[column].parse().ok())
                    .collect();
                assert_eq!(exported, shown, "export uses the same replay cutoff");
            }
            shown
        };
        let minute = |m: i64| DateTime::UNIX_EPOCH + Duration::minutes(m);
        chart.set_replay_time(Some(minute(90)));
        assert!(chart.set_replay_source(
            bars(&[1.0; 180], 1),
            crate::model::BarClock::new(Timeframe::Min1)
        ));
        // The other symbol's hour from 01:00 is unfinished at 01:30, so it has no value there.
        assert_eq!(values(&mut chart), [Some(10.0), None]);
        chart.set_replay_time(Some(minute(150)));
        assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), None]);
        chart.set_replay_time(Some(minute(90)));
        assert_eq!(values(&mut chart), [Some(10.0), None]);
        chart.set_replay_time(None);
        assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), Some(24.0)]);

        let driver = Arc::new(
            crate::data::ReplayDriver::new(
                (0..180)
                    .map(|m| {
                        crate::data::ReplayEvent::new(
                            chart.symbol(),
                            minute(m),
                            m as u64,
                            crate::data::MarketEvent::Trade(crate::model::Tick::new(
                                minute(m),
                                1.0,
                                1.0,
                                None,
                            )),
                        )
                    })
                    .collect(),
                vec![crate::model::Instrument::new(chart.symbol())],
            )
            .unwrap(),
        );
        driver.seek(minute(90));
        chart.set_market_replay(Some(driver.clone()));
        assert_eq!(values(&mut chart), [Some(10.0), None]);
        driver.seek(minute(150));
        assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), None]);
        driver.seek(minute(90));
        assert_eq!(values(&mut chart), [Some(10.0), None]);
        chart.set_market_replay(None);
        assert_eq!(values(&mut chart), [Some(10.0), Some(14.0), Some(24.0)]);
    }

    /// A volume profile on a finer interval loads that history back to the first bar its
    /// profiles in view read, not to the chart's first bar, and loads more only when that bar
    /// moves earlier. Price-based chart types keep loading to the chart's first bar.
    #[test]
    fn finer_profile_history_stops_at_the_profile_span() {
        use crate::model::ChartType;
        let feed = Arc::new(RetryFeed::default());
        let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 10], 60)));
        chart.set_symbol("MAIN");
        chart.set_timeframe(Timeframe::Hour1);
        let mut profile: Box<dyn crate::studies::Indicator> =
            Box::new(crate::studies::VolProfileVisible::new(10));
        profile
            .set_input(
                "source_interval",
                InputValue::Timeframe(Some(Timeframe::Min30)),
            )
            .unwrap();
        let id = chart.add_indicator(profile);
        let ctx = egui::Context::default();
        // Half-hour bars from 00:00 to 09:30, answered two at a time from the newest.
        let fine = bars(&[1.0; 20], 30);
        let hour = |hour: i64| Some(DateTime::UNIX_EPOCH + Duration::hours(hour));
        for (chart_type, view, expected) in [
            (ChartType::Candles, 8..10, vec![None, hour(9)]),
            (ChartType::Candles, 5..10, vec![hour(8), hour(7), hour(6)]),
            (ChartType::Candles, 9..10, vec![]),
            (
                ChartType::Renko,
                9..10,
                vec![hour(5), hour(4), hour(3), hour(2), hour(1)],
            ),
        ] {
            chart.set_chart_type(chart_type);
            chart
                .indicator_mut(id)
                .unwrap()
                .set_visible_range(view.clone());
            let mut befores = Vec::new();
            loop {
                ctx.run_ui(egui::RawInput::default(), |_| {})
                    .drop_without_applying_deltas();
                chart.pump_study_data(&ctx, Some(feed.clone()));
                let resolve = feed.resolves.lock().unwrap().pop();
                if let Some(reply) = resolve {
                    reply.send(Ok(Instrument::new("MAIN")));
                    continue;
                }
                let Some((request, reply)) = feed.histories.lock().unwrap().pop() else {
                    break;
                };
                befores.push(request.before);
                let end = fine
                    .partition_point(|bar| request.before.is_none_or(|before| bar.time < before));
                reply.send(Ok(HistoryPage::new(
                    fine[end.saturating_sub(2)..end].to_vec(),
                    end <= 2,
                )));
            }
            assert_eq!(befores, expected, "{chart_type:?} {view:?}");
        }
    }

    /// Guards #1667 and #1681: each full recompute of the profiles that read histories runs on
    /// its own frame, and a profile waiting for its turn keeps requesting the finer history it
    /// read and the start it read it from. So that history is loaded once from the newest bar,
    /// only back to the earliest profile start in view (the session open at 1970-01-13 00:00, or
    /// bar 305 at 17:00 for the visible range alone), and the recompute reads all 190 half-hour
    /// bars of volume 1 under bars 305 to 399. The feed answers ten hours a page.
    #[test]
    fn waiting_profiles_keep_their_finer_history() {
        use crate::model::ProfileResolution;
        use crate::studies::{Indicator, VolProfileSession, VolProfileVisible};
        let hour = |hour: i64| DateTime::UNIX_EPOCH + Duration::hours(hour);
        let hourly = bars(&[1.0; 400], 60);
        let suggested = |intervals: &[(Timeframe, Option<i64>)]| {
            Instrument::new("MAIN").with_profile_resolutions(intervals.iter().map(
                |&(timeframe, days)| {
                    ProfileResolution::new(timeframe, days.map(chrono::TimeDelta::days))
                },
            ))
        };
        let explicit = |mut study: Box<dyn Indicator>, timeframe| {
            study
                .set_input("source_interval", InputValue::Timeframe(Some(timeframe)))
                .unwrap();
            study
        };
        let session = || Box::new(VolProfileSession::default()) as Box<dyn Indicator>;
        let visible = || Box::new(VolProfileVisible::default()) as Box<dyn Indicator>;
        let finer = [(Timeframe::Min15, Some(1)), (Timeframe::Min30, None)];
        let half = [(Timeframe::Min30, None)];
        for (name, instrument, studies, main_page, last) in [
            (
                "session on 15 minutes, visible range on 30, as suggested",
                suggested(&finer),
                [session(), visible()],
                400,
                22,
            ),
            (
                "older chart bars arrive after the half-hour history",
                suggested(&half),
                [Box::new(SMA::new(5)) as Box<dyn Indicator>, visible()],
                200,
                22,
            ),
            (
                "both on suggested 30 minutes",
                suggested(&half),
                [session(), visible()],
                400,
                2,
            ),
            (
                "both on explicit 30 minutes",
                suggested(&[]),
                [
                    explicit(session(), Timeframe::Min30),
                    explicit(visible(), Timeframe::Min30),
                ],
                400,
                2,
            ),
            (
                "explicit 15 minute session, 30 minute visible range",
                suggested(&[]),
                [
                    explicit(session(), Timeframe::Min15),
                    explicit(visible(), Timeframe::Min30),
                ],
                400,
                22,
            ),
        ] {
            let feed = Arc::new(RetryFeed::default());
            let mut chart = ChartBuilder::new()
                .with_feed(feed.clone())
                .with_symbol("MAIN")
                .with_timeframe(Timeframe::Hour1)
                .build();
            let id = studies.map(|study| chart.add_indicator(study))[1];
            let ctx = egui::Context::default();
            let mut befores = Vec::new();
            for frame in 0..150 {
                shapes(&ctx, &mut chart, frame as f64, Vec::new());
                for reply in feed.resolves.lock().unwrap().drain(..) {
                    reply.send(Ok(instrument.clone()));
                }
                let pending: Vec<_> = feed.histories.lock().unwrap().drain(..).collect();
                for (request, reply) in pending {
                    let minutes = request.timeframe.duration_ms() / 60_000;
                    let before = request.before.unwrap_or(hour(400));
                    if minutes == 60 {
                        // Older chart bars are held until the half-hour history has loaded.
                        if request.before.is_some() && frame < 60 {
                            feed.histories.lock().unwrap().push((request, reply));
                            continue;
                        }
                        let end = hourly.partition_point(|bar| bar.time < before);
                        let start = end.saturating_sub(main_page);
                        let page = hourly[start..end].to_vec();
                        reply.send(Ok(HistoryPage::new(page, start == 0)));
                        continue;
                    }
                    if minutes == 30 {
                        befores.push(request.before);
                    }
                    let page: Vec<_> = bars(&vec![1.0; 24_000 / minutes as usize], minutes)
                        .into_iter()
                        .filter(|bar| bar.time < before && bar.time >= before - Duration::hours(10))
                        .collect();
                    let end = page.first().is_none_or(|bar| bar.time == hour(0));
                    reply.send(Ok(HistoryPage::new(page, end)));
                }
            }
            let held = chart
                .study_data
                .series
                .iter()
                .find(|series| series.request.timeframe == Some(Timeframe::Min30))
                .map_or(0, |series| {
                    series
                        .bars
                        .iter()
                        .filter(|bar| bar.time >= hour(305))
                        .count()
                });
            let profile = chart.indicators().find(|(key, _)| *key == id).unwrap().1;
            let volume: f64 = profile
                .volume_profiles(305..400)
                .iter()
                .flat_map(|plot| plot.profile.volume())
                .sum();
            assert_eq!(
                (
                    befores.iter().filter(|before| before.is_none()).count(),
                    befores.last().copied().flatten(),
                    held,
                    volume,
                ),
                (1, Some(hour(12 * 24 + last)), 190, 190.0),
                "{name}"
            );
        }
    }

    /// Guards #1668: the pass that runs every frame to resolve the study histories also works
    /// out how far back each loads, allocating for the requests alone. Two profiles read one
    /// finer history here; the session profile reads the half-hour bars from 00:00.
    #[test]
    fn study_data_reaches_allocate_only_the_requests() {
        use crate::studies::{Indicator, VolProfileSession, VolProfileVisible};
        let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 30], 60)));
        chart.set_symbol("MAIN");
        chart.set_timeframe(Timeframe::Hour1);
        for mut profile in [
            Box::new(VolProfileSession::new(10)) as Box<dyn Indicator>,
            Box::new(VolProfileVisible::new(10)),
        ] {
            profile
                .set_input(
                    "source_interval",
                    InputValue::Timeframe(Some(Timeframe::Min30)),
                )
                .unwrap();
            chart.add_indicator(profile);
        }
        let ctx = egui::Context::default();
        for frame in 0..3 {
            shapes(&ctx, &mut chart, frame as f64, Vec::new());
        }
        let (requests, expected) = allocations(|| chart.study_data_requests());
        let (reaches, resolved) = allocations(|| chart.study_data_reaches());
        assert_eq!(reaches, requests);
        assert_eq!(
            resolved,
            expected
                .into_iter()
                .map(|request| (request, Some(DateTime::UNIX_EPOCH)))
                .collect::<Vec<_>>()
        );
    }

    /// Drawings request one finer history and page only to their earliest visible anchor.
    #[cfg(feature = "drawings")]
    #[test]
    fn drawing_profile_history_stops_at_the_profile_span() {
        use crate::drawings::{ChartPoint, Drawing, DrawingToolType};
        use crate::model::{ChartType, ProfileResolution};
        use crate::studies::Indicator;
        let feed = Arc::new(RetryFeed::default());
        let mut chart = Chart::new(crate::model::BarData::from_bars(bars(&[1.0; 10], 60)));
        chart.set_symbol("MAIN");
        chart.set_timeframe(Timeframe::Hour1);
        chart.set_instrument(Some(Instrument::new("MAIN").with_profile_resolutions([
            ProfileResolution::new(Timeframe::Min30, Some(Duration::hours(12))),
        ])));
        let mut drawing = Drawing::new(1, DrawingToolType::FixedRangeVolumeProfile);
        drawing.chart_points = vec![ChartPoint::new(7.0, 1.0), ChartPoint::new(8.0, 1.0)];
        chart.drawings_mut().drawings.push(drawing);
        let request = StudyDataRequest {
            symbol: "MAIN".into(),
            timeframe: Some(Timeframe::Min30),
            metric: None,
        };
        assert_eq!(
            chart.study_data_requests().as_slice(),
            std::slice::from_ref(&request)
        );
        let ctx = egui::Context::default();
        let fine = bars(&[1.0; 20], 30);
        let hour = |h| Some(DateTime::UNIX_EPOCH + Duration::hours(h));
        for (anchor, expected) in [
            (7.0, vec![None, hour(9), hour(8)]),
            (5.0, vec![hour(7), hour(6)]),
        ] {
            chart.drawings_mut().drawings[0].chart_points[0].bar_idx = anchor;
            let mut befores = Vec::new();
            loop {
                ctx.run_ui(egui::RawInput::default(), |_| {})
                    .drop_without_applying_deltas();
                chart.pump_study_data(&ctx, Some(feed.clone()));
                if let Some(reply) = feed.resolves.lock().unwrap().pop() {
                    reply.send(Ok(Instrument::new("MAIN")));
                    continue;
                }
                let Some((asked, reply)) = feed.histories.lock().unwrap().pop() else {
                    break;
                };
                befores.push(asked.before);
                let end =
                    fine.partition_point(|bar| asked.before.is_none_or(|before| bar.time < before));
                reply.send(Ok(HistoryPage::new(
                    fine[end.saturating_sub(2)..end].to_vec(),
                    end <= 2,
                )));
            }
            assert_eq!(befores, expected, "anchor {anchor}");
        }
        // A study sharing the interval still controls its own earlier reach.
        let mut study = crate::studies::VolProfileFixed::new(2);
        study
            .set_input(
                "source_interval",
                InputValue::Timeframe(Some(Timeframe::Min30)),
            )
            .unwrap();
        study.set_timeframe(Timeframe::Hour1);
        study.calculate(&bars(&[1.0; 10], 60));
        study.set_visible_range(0..10);
        let id = chart.add_indicator(Box::new(study));
        assert_eq!(chart.study_data_requests(), [request]);
        assert_eq!(chart.study_data_reaches()[0].1, hour(0));
        chart.remove_series(id);
        chart.drawings_mut().drawings[0].visible = false;
        assert!(chart.study_data_requests().is_empty());
        chart.drawings_mut().drawings[0].visible = true;
        chart.set_chart_type(ChartType::Renko);
        assert!(chart.study_data_requests().is_empty());
    }
}