egui_charts/widget/feed/
mod.rs

1//! A chart driven by a [`Datafeed`]: requests issued from `show`, answers and live pushes applied
2//! once per frame.
3//!
4//! Every feed call but the one in [`FeedLink`]'s `Drop` comes from [`Chart::pump`], which runs
5//! outside any egui closure: a feed may answer or push inline, and that runs the waker, which
6//! takes the egui context's lock.
7
8mod pump;
9
10use super::{Chart, SeriesId};
11use crate::chart::overlays::{VolumeBubbles, VolumeBubblesConfig};
12use crate::data::inbox::Inbox;
13use crate::data::trades::Trades;
14use crate::data::{
15    Datafeed, DepthHistory, DepthRequest, FeedCapabilities, FeedError, Footprint, HistoryRequest,
16    Reply, SubscriptionId, TradeRequest,
17};
18use crate::model::{BarClock, Instrument, OrderBook, Tick, Timeframe};
19use crate::theme::ThemeExt;
20use egui::ViewportId;
21use std::sync::{Arc, Mutex, PoisonError};
22use std::time::Duration;
23
24/// Trades kept per series for order flow.
25pub(crate) const TRADE_CAPACITY: usize = 100_000;
26/// Bars a feed-backed footprint keeps, the oldest dropped first.
27const FOOTPRINT_BARS: usize = 5_000;
28/// Price rows a feed-backed footprint keeps in all its bars, the oldest bars dropped first.
29const FOOTPRINT_ROWS: usize = TRADE_CAPACITY / 2;
30
31/// Bars from the first visible bar to the oldest one below which an older page is asked for.
32const SCROLL_BACK_MARGIN: usize = 150;
33
34/// The smallest scroll-back page asked for.
35const MIN_SCROLL_BACK_PAGE: usize = 600;
36
37/// A chart's link to its feed.
38pub(crate) struct FeedLink {
39    pub(super) feed: Arc<dyn Datafeed>,
40    inbox: Arc<Inbox>,
41    /// Slots of the chart's timeframe, by which history, live bars and ticks merge.
42    pub(super) clock: BarClock,
43    initial_bars: usize,
44    /// Whether history is requested adjusted for dividends.
45    pub(super) adjusted: bool,
46    /// The viewport the chart was last shown in, which the waker wakes. Set by the first `pump`,
47    /// which installs the waker, and cleared by `Drop` before it unsubscribes, so the waker does
48    /// nothing then.
49    wake: Arc<Mutex<Option<ViewportId>>>,
50    /// The symbol or timeframe changed: the next `pump` reloads.
51    reload_pending: bool,
52    pub(super) paused: bool,
53    /// The pause last set with [`Chart::set_feed_paused`]. `paused` also holds while the main
54    /// series is removed or a replay runs.
55    pub(super) host_paused: bool,
56    next_key: u64,
57    /// Requests not answered yet; an answer under any other key is stale and dropped.
58    pending: Vec<(u64, Request, u32)>,
59    retries: Vec<(f64, Request, u32)>,
60    subscription: Option<SubscriptionId>,
61    instrument: Option<Instrument>,
62    /// What the feed offers as a whole, once it has answered.
63    capabilities: Option<FeedCapabilities>,
64    pub(super) trades: Trades,
65    /// The held trades as volume bubbles, when the host turned them on.
66    pub(super) bubbles: Option<VolumeBubbles>,
67    pub(super) bubbles_config: VolumeBubblesConfig,
68    /// The trades as a footprint on the instrument's tick, while the chart type or a volume
69    /// profile drawing reads them, within [`FOOTPRINT_BARS`] and [`FOOTPRINT_ROWS`].
70    pub(super) footprint: Option<Footprint>,
71    /// The footprint of the trades before the replay cutoff, while a replay runs under a visible
72    /// volume profile drawing, rebuilt when the cutoff moves.
73    pub(super) replay_footprint: Option<(chrono::DateTime<chrono::Utc>, Option<Footprint>)>,
74    order_book: Option<OrderBook>,
75    order_queue: Option<crate::data::OrderQueue>,
76    /// The order book over time, when the host turned it on.
77    pub(super) depth_history: Option<DepthHistory>,
78    depth_complete: bool,
79    /// Where older history continues after an empty page, by source: bars, trades, depth. The
80    /// bar cursor keeps the oldest bar held when the feed named it and lapses once that bar goes.
81    bar_cursor: Option<(
82        chrono::DateTime<chrono::Utc>,
83        Option<chrono::DateTime<chrono::Utc>>,
84    )>,
85    trade_cursor: Option<chrono::DateTime<chrono::Utc>>,
86    depth_cursor: Option<chrono::DateTime<chrono::Utc>>,
87    /// No older trades remain to be loaded.
88    trades_complete: bool,
89    /// A trade page reported that nothing exists before its oldest trade, or came back empty
90    /// without `continue_before`: no trade before the oldest held one was missed.
91    trades_from_start: bool,
92    pub(super) heatmap: Option<std::cell::RefCell<crate::chart::overlays::LiquidityHeatmap>>,
93    heatmap_config: crate::chart::overlays::LiquidityHeatmapConfig,
94    /// How the live book is shown by the heatmap's profile and the depth ladder.
95    book_view: crate::chart::overlays::OrderBookView,
96    /// None keeps the profile following the heatmap for existing setter callers.
97    pub(super) book_visible: Option<bool>,
98    study_inputs: Vec<(crate::data::OrderFlowInput, u64)>,
99    time: Option<chrono::DateTime<chrono::Utc>>,
100    /// A permanent history or resolve failure stops paging until the next reload.
101    pub(super) paging_stopped: bool,
102}
103
104#[derive(Clone)]
105enum Request {
106    Capabilities,
107    Resolve,
108    History {
109        request: HistoryRequest,
110        scroll_back: bool,
111    },
112    Trades(TradeRequest),
113    Depth(DepthRequest),
114}
115
116impl FeedLink {
117    pub(super) fn server_time(&self) -> Option<chrono::DateTime<chrono::Utc>> {
118        self.inbox.server_time(self.subscription?)
119    }
120
121    /// A reply for `request`, under a new key.
122    fn reply<T: Send + 'static>(&mut self, request: Request) -> Reply<T> {
123        self.next_key += 1;
124        self.pending.push((self.next_key, request, 0));
125        self.inbox.reply(self.next_key)
126    }
127
128    /// Sums the held trades again, after the clock, the instrument or the trades changed.
129    fn rebuild_bubbles(&mut self) {
130        if let Some(bubbles) = self.bubbles.take() {
131            self.bubbles = Some(self.bubbles(bubbles.config));
132        }
133        if self.footprint.is_some() {
134            self.footprint = self.footprint(None);
135        }
136    }
137
138    /// A paused link for an offscreen image: it holds the visuals the feed built, the footprint,
139    /// bubbles, order book and depth, and never talks to the feed or the live link's inbox.
140    pub(super) fn frozen(&self) -> Self {
141        Self {
142            feed: Arc::clone(&self.feed),
143            inbox: Arc::default(),
144            clock: self.clock.clone(),
145            initial_bars: self.initial_bars,
146            adjusted: self.adjusted,
147            wake: Arc::default(),
148            reload_pending: false,
149            paused: true,
150            host_paused: true,
151            next_key: 0,
152            pending: Vec::new(),
153            retries: Vec::new(),
154            subscription: None,
155            instrument: self.instrument.clone(),
156            capabilities: self.capabilities.clone(),
157            trades: Trades::new(TRADE_CAPACITY),
158            bubbles: self
159                .bubbles
160                .as_ref()
161                .map(|bubbles| self.bubbles(bubbles.config.clone())),
162            footprint: self
163                .footprint
164                .is_some()
165                .then(|| self.footprint(None))
166                .flatten(),
167            replay_footprint: self.replay_footprint.clone(),
168            order_book: self.order_book.clone(),
169            order_queue: self.order_queue.clone(),
170            depth_history: self.depth_history.clone(),
171            depth_complete: self.depth_complete,
172            bar_cursor: None,
173            trade_cursor: None,
174            depth_cursor: None,
175            trades_complete: false,
176            trades_from_start: false,
177            heatmap: self.heatmap.is_some().then(Default::default),
178            heatmap_config: self.heatmap_config,
179            book_view: self.book_view.clone(),
180            book_visible: self.book_visible,
181            bubbles_config: self.bubbles_config.clone(),
182            study_inputs: Vec::new(),
183            time: self.time,
184            paging_stopped: true,
185        }
186    }
187
188    /// The held trades as a footprint, or `None` without the instrument's tick size. A trade
189    /// older than the one before it is left out, as it is when it arrives live. The bar of the
190    /// oldest held trade is incomplete until a trade page reports the start of trade history.
191    /// With `before`, includes only executions strictly before that replay cutoff.
192    pub(super) fn footprint(
193        &self,
194        before: Option<chrono::DateTime<chrono::Utc>>,
195    ) -> Option<Footprint> {
196        let tick_size = self.instrument.as_ref()?.tick_size?;
197        let mut flow = Footprint::new(self.clock.clone(), tick_size, 1).ok()?;
198        flow.set_budget(FOOTPRINT_BARS, FOOTPRINT_ROWS);
199        let trades = || {
200            self.trades
201                .iter()
202                .take_while(|trade| before.is_none_or(|time| trade.ts < time))
203        };
204        for trade in trades() {
205            let _ = flow.push(trade);
206        }
207        if let Some(lost) = self.trades.lost_until() {
208            flow.drop_through(lost, trades());
209        }
210        if let Some(first) = self.unconfirmed_start() {
211            flow.received_from(first);
212        }
213        Some(flow)
214    }
215
216    /// The oldest held trade, where the footprint starts, while executions before it may
217    /// exist that never arrived: no trade page reported the start of trade history and the
218    /// store lost none. A loss is marked exactly by `drop_through` instead.
219    fn unconfirmed_start(&self) -> Option<chrono::DateTime<chrono::Utc>> {
220        if self.trades_from_start || self.trades.lost_until().is_some() {
221            return None;
222        }
223        self.trades.iter().next().map(|trade| trade.ts)
224    }
225
226    fn bubbles(&self, config: VolumeBubblesConfig) -> VolumeBubbles {
227        let tick_size = self.instrument.as_ref().and_then(|i| i.tick_size);
228        VolumeBubbles::new(config, self.clock.clone(), tick_size, self.trades.iter())
229    }
230}
231
232impl Drop for FeedLink {
233    fn drop(&mut self) {
234        *lock(&self.wake) = None;
235        if let Some(id) = self.subscription.take() {
236            self.inbox.unsubscribe(id);
237            self.feed.unsubscribe(id);
238        }
239    }
240}
241
242impl Chart {
243    /// Drives the chart from `feed` from the next [`show`](Self::show) on.
244    pub(crate) fn attach_feed(&mut self, feed: Arc<dyn Datafeed>, initial_bars: usize) {
245        self.set_timeframe(self.timeframe.unwrap_or_default());
246        let timeframe = self.timeframe.unwrap_or_default();
247        self.feed_link = Some(FeedLink {
248            feed,
249            inbox: Arc::default(),
250            clock: BarClock::new(timeframe),
251            initial_bars,
252            adjusted: false,
253            wake: Arc::default(),
254            reload_pending: true,
255            paused: false,
256            host_paused: false,
257            next_key: 0,
258            pending: Vec::new(),
259            retries: Vec::new(),
260            subscription: None,
261            instrument: None,
262            capabilities: None,
263            trades: Trades::new(TRADE_CAPACITY),
264            bubbles: None,
265            footprint: None,
266            replay_footprint: None,
267            order_book: None,
268            order_queue: None,
269            depth_history: None,
270            depth_complete: false,
271            bar_cursor: None,
272            trade_cursor: None,
273            depth_cursor: None,
274            trades_complete: false,
275            trades_from_start: false,
276            heatmap: None,
277            heatmap_config: Default::default(),
278            book_view: Default::default(),
279            book_visible: None,
280            bubbles_config: Default::default(),
281            study_inputs: Vec::new(),
282            time: None,
283            paging_stopped: false,
284        });
285    }
286
287    /// What the feed offers as a whole, once it has answered [`Datafeed::capabilities`]. Pass it
288    /// to the interval and symbol selectors before a symbol resolves. `None` without a feed or
289    /// before the answer.
290    pub fn feed_capabilities(&self) -> Option<&FeedCapabilities> {
291        self.feed_link.as_ref()?.capabilities.as_ref()
292    }
293
294    /// Suspends feed requests and unsubscribes while the host supplies replay bars by hand.
295    /// Resuming reloads the current symbol and interval, including any trades retained by the
296    /// host. Bars supplied while paused are matched by exact opening time.
297    /// The feed also stays paused while the main series is removed, and follows this setting
298    /// again once undo restores the series.
299    pub fn set_feed_paused(&mut self, paused: bool) {
300        if let Some(link) = &mut self.feed_link {
301            link.host_paused = paused;
302        }
303        let paused = paused || !self.main_series_present || self.replay.is_some();
304        if paused {
305            self.study_data.pause();
306        }
307        if let Some(link) = &mut self.feed_link
308            && link.paused != paused
309        {
310            link.paused = paused;
311            link.reload_pending = true;
312            if paused {
313                if let Some(id) = link.subscription.take() {
314                    link.inbox.unsubscribe(id);
315                    link.feed.unsubscribe(id);
316                }
317                link.pending.clear();
318            } else {
319                self.study_data.clear_feed();
320            }
321        }
322    }
323
324    /// Shows `symbol` from the next [`show`](Self::show). With a feed, a changed symbol clears
325    /// the series and its studies, then resolves, loads and subscribes it, in one load per
326    /// `show` however many setters ran; an unchanged one does nothing, also after an error.
327    /// Drawings and markers are retained separately for each symbol. A changed symbol clears
328    /// zoom history and returns its price scale to autoscaling. It also clears chart undo
329    /// history, unless the host makes it inside [`Self::user_action`]: then the navigation is
330    /// an undo step that restores the symbol, interval and viewport, and the steps before it
331    /// stay with their own symbol. Without a feed replace the bars with
332    /// [`set_data`](Self::set_data).
333    pub fn set_symbol(&mut self, symbol: impl Into<String>) {
334        let keep_history = self.history.recording();
335        self.change_symbol(symbol.into(), keep_history);
336    }
337
338    /// Shows `symbol` from the next frame, keeping chart undo history when asked to.
339    pub(super) fn change_symbol(&mut self, symbol: String, keep_history: bool) {
340        if symbol != self.symbol {
341            self.clear_source_signals();
342            self.reload_price_targets();
343            self.set_replay_time(None);
344        }
345        if symbol != self.symbol
346            && let Some(link) = &mut self.feed_link
347        {
348            link.reload_pending = true;
349            link.instrument = None;
350        }
351        if symbol != self.symbol {
352            self.host_events
353                .push(super::ChartEvent::SymbolChanged(symbol.clone()));
354            #[cfg(feature = "drawings")]
355            self.anchor_drawing_points();
356            #[cfg(feature = "drawings")]
357            {
358                self.save_stored_drawings();
359                self.drawings.cancel_curr_drawing();
360                self.drawings.end_drag_handle();
361                self.drawings.deselect();
362                self.detach_pending_drawings();
363                let mut next = self.symbol_drawings.remove(&symbol).unwrap_or_default();
364                next.0.set_sync_enabled(self.drawings.sync_enabled);
365                next.0.registry = self.drawings.registry.clone();
366                next.0.options = self.drawings.options.clone();
367                next.0.tool_defaults = self.drawings.tool_defaults.clone();
368                next.0.set_active_tool(self.drawings.active_tool);
369                if let Some(id) = self.drawings.active_custom_tool() {
370                    let _ = next.0.set_custom_tool(id);
371                }
372                next.0.stay_in_drawing_mode = self.drawings.stay_in_drawing_mode;
373                next.0.eraser_mode = self.drawings.eraser_mode;
374                next.0.curr_timeframe = self.drawings.curr_timeframe.clone();
375                std::mem::swap(&mut self.drawings, &mut next.0);
376                std::mem::swap(&mut self.marks, &mut next.1);
377                self.attach_pending_drawings();
378                if !next.0.drawings.is_empty()
379                    || !next.1.is_empty()
380                    || !next
381                        .0
382                        .pending_sync
383                        .lock()
384                        .unwrap_or_else(PoisonError::into_inner)
385                        .is_empty()
386                {
387                    self.symbol_drawings.insert(self.symbol.clone(), next);
388                }
389            }
390            #[cfg(not(feature = "drawings"))]
391            {
392                let mut next_marks = self.symbol_markers.remove(&symbol).unwrap_or_default();
393                std::mem::swap(&mut self.marks, &mut next_marks);
394                if !next_marks.is_empty() {
395                    self.symbol_markers.insert(self.symbol.clone(), next_marks);
396                }
397            }
398            if !keep_history {
399                self.history.clear();
400            }
401            self.history.edits += 1;
402            self.state.price_scale_mut().reset();
403            self.state.clear_zoom_history();
404            self.pending_scroll_time = None;
405            self.legend_instrument = None;
406            self.currency_choice = None;
407            self.bar_close_time = None;
408            self.server_time_offset = None;
409            self.legend_badge = None;
410            self.legend_statuses.clear();
411            self.quote = None;
412            self.quote_action = None;
413            self.session_quote = None;
414            self.events = Default::default();
415            self.time_scale_events = Default::default();
416            self.event_click = None;
417            self.axis_hover = None;
418            self.state.price_scale_mut().set_price_precision(None);
419            self.state.price_scale_mut().set_tick_size(None);
420        }
421        self.symbol = symbol;
422    }
423
424    /// Drops the cached drawings, markers and storage reader of an inactive symbol.
425    /// Returns `false` for the current symbol or while local drawing edits await a storage
426    /// read or write acknowledgement. Otherwise returns `true`, also for an uncached symbol.
427    /// Stored documents are left in host storage and read again on the next visit. Without
428    /// storage, forgotten drawings and markers are lost, including pending sync notifications.
429    pub fn forget_symbol(&mut self, symbol: &str) -> bool {
430        if symbol == self.symbol {
431            return false;
432        }
433        #[cfg(feature = "drawings")]
434        {
435            if let Some(store) = &mut self.drawing_store
436                && !store.forget(symbol)
437            {
438                return false;
439            }
440            self.symbol_drawings.remove(symbol);
441            self.symbol_drawings.shrink_to_fit();
442        }
443        #[cfg(not(feature = "drawings"))]
444        {
445            self.symbol_markers.remove(symbol);
446            self.symbol_markers.shrink_to_fit();
447        }
448        true
449    }
450
451    /// Shows bars of `timeframe` from the next [`show`](Self::show), by the rules of
452    /// [`set_symbol`](Self::set_symbol). Without a feed it sets the legend label and the
453    /// session breaks only. It is also the interval the drawings' per-interval visibility is
454    /// checked against.
455    pub fn set_timeframe(&mut self, timeframe: Timeframe) {
456        if self.timeframe != Some(timeframe) {
457            self.clear_source_signals();
458            self.host_events
459                .push(super::ChartEvent::TimeframeChanged(timeframe));
460            self.bar_close_time = None;
461        }
462        if self.timeframe != Some(timeframe)
463            && let Some(link) = &mut self.feed_link
464        {
465            link.reload_pending = true;
466        }
467        let changed = self.timeframe != Some(timeframe);
468        if changed {
469            self.pending_scroll_time = None;
470        }
471        self.timeframe = Some(timeframe);
472        for study in self.studies.indicators_mut() {
473            study.set_timeframe(timeframe);
474        }
475        #[cfg(feature = "drawings")]
476        {
477            self.drawings.curr_timeframe = timeframe.as_str().into_owned();
478        }
479        if changed {
480            self.pending_from.fill(Some(0));
481            self.next_refresh.fill(f64::NEG_INFINITY);
482            self.element_studies.dirty = true;
483            self.apply_replay_window();
484        }
485    }
486
487    /// Requests regular hours (`false`, the default) or regular and extended hours (`true`).
488    /// A changed choice reloads history and replaces the live subscription on the next frame.
489    /// Without a feed the host supplies the selected bars with `set_data`.
490    pub fn set_extended_hours(&mut self, extended_hours: bool) {
491        if self.extended_hours != extended_hours {
492            self.extended_hours = extended_hours;
493            self.study_data.clear_feed();
494            if let Some(link) = &mut self.feed_link {
495                link.reload_pending = true;
496            }
497        }
498    }
499
500    /// Asks the feed for data in a currency or unit the host chose, such as `"EUR"` or `"oz"`,
501    /// or in the instrument's own with `None`. The library converts nothing: the feed does, and
502    /// labels the resolved instrument's currency and unit, which the legend and scale headers
503    /// show. A changed choice drops the bars, trades and depth held, ignores the replies to the
504    /// previous choice and reloads, resolving through [`Datafeed::resolve_converted`] and
505    /// carrying the choice in every history, trade and depth request and subscription,
506    /// including those of studies on other symbols. A feed that cannot convert answers
507    /// [`FeedError::NotSupported`] and the chart shows no data. A feed attached later uses it.
508    pub fn set_conversion(&mut self, conversion: Option<String>) {
509        if self.conversion != conversion {
510            self.conversion = conversion;
511            self.study_data.clear_feed();
512            if let Some(link) = &mut self.feed_link {
513                link.reload_pending = true;
514                // The headers must not keep the previous choice's currency while loading.
515                link.instrument = None;
516            }
517        }
518    }
519
520    /// The currency or unit chosen with [`set_conversion`](Self::set_conversion).
521    pub fn conversion(&self) -> Option<&str> {
522        self.conversion.as_deref()
523    }
524
525    /// Requests history adjusted for dividends (`true`) or as traded (`false`, the default), when
526    /// the feed supports it; comparisons are requested the same way. A changed choice reloads
527    /// history and replaces the live subscription on the next frame. Without a feed the host
528    /// supplies the selected bars with `set_data`.
529    pub fn set_dividend_adjusted(&mut self, adjusted: bool) {
530        if let Some(link) = &mut self.feed_link
531            && link.adjusted != adjusted
532        {
533            link.adjusted = adjusted;
534            link.reload_pending = true;
535            self.study_data.clear_feed();
536        }
537    }
538
539    /// Whether history is requested adjusted for dividends.
540    pub fn dividend_adjusted(&self) -> bool {
541        self.feed_link.as_ref().is_some_and(|link| link.adjusted)
542    }
543
544    /// Reloads the current feed key on the next frame, retaining drawings and chart history.
545    /// Use after changing host-side feed settings or to retry a permanent request failure.
546    pub fn reload_feed(&mut self) {
547        self.study_data.clear_feed();
548        if let Some(link) = &mut self.feed_link {
549            link.reload_pending = true;
550        }
551    }
552
553    /// Whether history and subscriptions request extended hours.
554    pub fn extended_hours(&self) -> bool {
555        self.extended_hours
556    }
557
558    /// The symbol the chart shows.
559    pub fn symbol(&self) -> &str {
560        &self.symbol
561    }
562
563    /// The timeframe the chart shows. A chart built with a feed always has one; `None` only for
564    /// a chart made with [`Chart::new`] until [`set_timeframe`](Self::set_timeframe) is called.
565    pub fn timeframe(&self) -> Option<Timeframe> {
566        self.timeframe
567    }
568
569    /// The feed's description of the symbol, or metadata supplied with `set_instrument`.
570    pub fn instrument(&self) -> Option<&Instrument> {
571        self.feed_link
572            .as_ref()
573            .and_then(|link| link.instrument.as_ref())
574            .or(self.legend_instrument.as_ref())
575    }
576
577    /// The trades of `series` the feed sent, oldest first, at most 100,000. Empty for any id but
578    /// [`SeriesId::MAIN`] and without a feed.
579    pub fn trades(&self, series: SeriesId) -> impl Iterator<Item = &Tick> {
580        let link = self.feed_link.as_ref().filter(|_| series == SeriesId::MAIN);
581        link.into_iter().flat_map(|link| link.trades.iter())
582    }
583
584    /// Combines the retained fills with explicit aggressor ids, in first-fill order. At most
585    /// 100,000 entries; fills without an id stay separate. Rebuilds the view on request, including
586    /// fills arriving in separate pages or frames. Per-price consumers keep using `trades`.
587    pub fn combined_trades(&self, series: SeriesId) -> Vec<crate::data::CombinedTrade> {
588        crate::data::combine_trades(self.trades(series))
589    }
590
591    /// Draws the trades the feed sends as bubbles, grouped and coloured as the config says,
592    /// from the next frame, or stops with `None`. By default, nearby bars and price levels merge
593    /// when crowded, and diameters adapt to the available screen space. A chart without a feed
594    /// holds no trades and draws none, and Renko, Kagi, line break, point and figure and range
595    /// charts draw none, since their x axis is not time. Sums every held trade again: call it
596    /// when the settings change, not every frame.
597    pub fn set_volume_bubbles(&mut self, config: Option<VolumeBubblesConfig>) {
598        if let Some(link) = &mut self.feed_link {
599            if let Some(config) = &config {
600                link.bubbles_config = config.clone();
601            }
602            link.bubbles = config.map(|config| link.bubbles(config));
603        }
604    }
605
606    /// The order book of `series` the feed sent, if it holds one. Unavailable during candle-only
607    /// replay; attach a market replay driver for recorded book state.
608    pub fn order_book(&self, series: SeriesId) -> Option<&OrderBook> {
609        let link = self
610            .feed_link
611            .as_ref()
612            .filter(|_| series == SeriesId::MAIN && self.replay.is_none())?;
613        link.order_book.as_ref()
614    }
615
616    /// Individual resting orders and their queue priority, when the feed sends order-level data.
617    /// Use `OrderQueue::aggregate` with a minimum size for filtered ladders or heatmaps, and
618    /// `largest_order` for the largest resting order at a level. Aggregated-only feeds return
619    /// `None`; a lost order book has no orders until a replacement snapshot arrives. Candle-only
620    /// replay hides the live queue.
621    pub fn order_queue(&self, series: SeriesId) -> Option<&crate::data::OrderQueue> {
622        self.feed_link
623            .as_ref()
624            .filter(|_| series == SeriesId::MAIN && self.replay.is_none())?
625            .order_queue
626            .as_ref()
627    }
628
629    /// How the heatmap's profile and the depth ladder show the live book: which individual orders
630    /// count, what a bar measures and which orders are highlighted. Its size tiers also mark the
631    /// heatmap's cells whose largest recorded order reaches one; the orders the heatmap records are
632    /// those of [`DepthHistoryConfig::orders`](crate::data::DepthHistoryConfig::orders). Applies
633    /// from the next frame. The profile key names its measure and maximum across all supplied
634    /// levels, including offscreen prices, after filtering; hovering the strip shows the row's
635    /// exact price, side and size. Set [`OrderBookView::quantity_unit`](crate::chart::overlays::OrderBookView::quantity_unit)
636    /// to describe the feed's quantities; without it the key says "feed units".
637    /// A view that needs individual orders on an aggregate feed draws an explicit notice in the
638    /// pane and an empty ladder, rather than aggregate sizes.
639    pub fn set_order_book_view(&mut self, view: crate::chart::overlays::OrderBookView) {
640        if let Some(link) = &mut self.feed_link {
641            link.book_view = view;
642        }
643    }
644
645    /// Show the live book profile at the right edge of the pane, independently of the heatmap.
646    /// Until explicitly set, the profile follows the heatmap for existing setter callers.
647    /// Requires a feed and the instrument's tick size; does nothing without a feed.
648    pub fn set_order_book_visible(&mut self, visible: bool) {
649        if let Some(link) = &mut self.feed_link {
650            link.book_visible = Some(visible);
651        }
652    }
653
654    /// The view set by [`set_order_book_view`](Self::set_order_book_view), the default of the
655    /// aggregate book without a feed.
656    pub fn order_book_view(&self) -> crate::chart::overlays::OrderBookView {
657        self.feed_link
658            .as_ref()
659            .map(|link| link.book_view.clone())
660            .unwrap_or_default()
661    }
662
663    /// Loads recorded depth from the feed and keeps `history` of its live book, or stops
664    /// with `None`. The history starts empty and is cleared when the symbol or interval changes.
665    /// It needs the instrument's tick size, and keeps no columns without one or without a feed.
666    /// Columns follow the times of the depth updates: an update stamped more than a minute after
667    /// the feed's time ([`Sink::set_server_time`](crate::data::Sink::set_server_time)) samples
668    /// nothing, so a wrong time cannot hold the history back.
669    /// The chart takes the columns sampled since its last frame when it is drawn: while it is not,
670    /// at most [`columns`](crate::data::DepthHistoryConfig::columns) of them wait, the oldest
671    /// making room for new ones.
672    pub fn set_depth_history(&mut self, history: Option<DepthHistory>) {
673        if let Some(link) = &mut self.feed_link {
674            if let Some(id) = link.subscription {
675                let config = history.as_ref().map(|history| history.config().clone());
676                link.inbox.sample_depth(id, config);
677            }
678            link.depth_history = history;
679            link.depth_complete = false;
680            link.depth_cursor = None;
681            link.pending
682                .retain(|(_, r, _)| !matches!(r, Request::Depth(_)));
683            link.retries
684                .retain(|(_, r, _)| !matches!(r, Request::Depth(_)));
685        }
686    }
687
688    /// Shows resting liquidity behind the price series. A feed and instrument tick size are
689    /// required. Enabling starts the default depth history if none is configured; disabling
690    /// releases textures and retains that history and the appearance settings. Re-enabling an
691    /// enabled heatmap keeps its textures. Trade bubbles and best bid/ask lines can be
692    /// enabled independently through the existing chart options.
693    pub fn set_liquidity_heatmap(&mut self, enabled: bool) {
694        if enabled
695            && self
696                .feed_link
697                .as_ref()
698                .is_none_or(|link| link.depth_history.is_none())
699        {
700            self.set_depth_history(Some(
701                DepthHistory::new(Default::default()).expect("default depth history"),
702            ));
703        }
704        if let Some(link) = &mut self.feed_link {
705            if enabled {
706                link.heatmap.get_or_insert_with(Default::default);
707            } else {
708                link.heatmap = None;
709            }
710        }
711    }
712
713    /// Sets the heatmap's cell brightness and resting-size range from the next frame. A changed
714    /// appearance redraws the held columns once; an unchanged effective setting uploads nothing.
715    /// Retained while disabled and in exported images. Does nothing without a feed; enable the
716    /// heatmap separately with [`set_liquidity_heatmap`](Self::set_liquidity_heatmap).
717    pub fn set_liquidity_heatmap_config(
718        &mut self,
719        config: crate::chart::overlays::LiquidityHeatmapConfig,
720    ) {
721        if let Some(link) = &mut self.feed_link {
722            link.heatmap_config = config;
723        }
724    }
725
726    /// The heatmap appearance supplied by the host, or its default without a feed.
727    pub fn liquidity_heatmap_config(&self) -> crate::chart::overlays::LiquidityHeatmapConfig {
728        self.feed_link
729            .as_ref()
730            .map(|link| link.heatmap_config)
731            .unwrap_or_default()
732    }
733
734    pub(super) fn paint_liquidity(
735        &self,
736        painter: &egui::Painter,
737        mapping: &crate::chart::coords::ChartMapping,
738    ) {
739        if self.replay.is_some() {
740            return;
741        }
742        let Some(link) = &self.feed_link else { return };
743        let Some(tick) = self.instrument().and_then(|i| i.tick_size) else {
744            return;
745        };
746        let mut unavailable = None;
747        if let (Some(heatmap), Some(history)) = (&link.heatmap, &link.depth_history) {
748            let mut heatmap = heatmap.borrow_mut();
749            heatmap.config = link.heatmap_config;
750            heatmap.render(
751                painter,
752                mapping,
753                history,
754                tick * history.config().ticks_per_row as f64,
755                |time| self.time_to_index(time).and_then(|i| self.index_to_x(i)),
756                &link.book_view,
757                [self.config.bullish_color, self.config.bearish_color],
758            );
759            unavailable = (history.config().orders.is_some() && link.order_queue.is_none())
760                .then_some(crate::data::OrderFilterUnavailable::AggregateDepth);
761        }
762        if link.book_visible.unwrap_or(link.heatmap.is_some()) {
763            unavailable = crate::chart::overlays::render_book_profile(
764                painter,
765                mapping,
766                link.order_book.as_ref(),
767                link.order_queue.as_ref(),
768                tick,
769                &link.book_view,
770                [self.config.bullish_color, self.config.bearish_color],
771            )
772            .err()
773            .or(unavailable);
774        }
775        if let Some(reason) = unavailable {
776            painter.text(
777                mapping.rect.right_top() + egui::vec2(-8.0, 8.0),
778                egui::Align2::RIGHT_TOP,
779                crate::localization::format_text(
780                    painter.ctx(),
781                    "Order-level depth unavailable: {0}",
782                    &[(
783                        "0",
784                        crate::localization::text(painter.ctx(), reason.to_string()),
785                    )],
786                ),
787                egui::FontId::proportional(11.0),
788                painter.ctx().active_theme().ui.warning,
789            );
790        }
791    }
792
793    /// The order book history of `series`, if [`set_depth_history`](Self::set_depth_history)
794    /// turned it on. Candle-only replay hides the live history.
795    pub fn depth_history(&self, series: SeriesId) -> Option<&DepthHistory> {
796        let link = self
797            .feed_link
798            .as_ref()
799            .filter(|_| series == SeriesId::MAIN && self.replay.is_none())?;
800        link.depth_history.as_ref()
801    }
802
803    /// Calendar boundaries selected by the chart, including the resolved exchange or local zone.
804    pub fn bar_clock(&self) -> BarClock {
805        let clock = BarClock::new(self.timeframe.unwrap_or_default());
806        let clock = match self.display_timezone() {
807            crate::scales::TimezoneMode::Local => clock.with_local_timezone(),
808            zone => clock.with_timezone(zone.to_tz().unwrap_or(crate::model::TimezoneId::utc())),
809        };
810        match self.instrument().and_then(|instrument| {
811            Some((
812                instrument
813                    .time_zone
814                    .clone()
815                    .unwrap_or_else(crate::model::TimezoneId::utc),
816                instrument.session.clone()?,
817            ))
818        }) {
819            Some((zone, session)) => clock.with_session(zone, session, self.extended_hours()),
820            None => clock,
821        }
822    }
823
824    /// The clock bars merge by: the feed's, or exact times without a feed.
825    pub(super) fn clock(&self) -> BarClock {
826        self.feed_link
827            .as_ref()
828            .filter(|link| !link.paused)
829            .map_or_else(BarClock::exact, |link| link.clock.clone())
830    }
831}
832
833/// Describes `symbol` as the feed does for `conversion`, or as it is without one.
834pub(super) fn resolve(
835    feed: &dyn Datafeed,
836    symbol: &str,
837    conversion: Option<&str>,
838    reply: Reply<Instrument>,
839) {
840    match conversion {
841        Some(conversion) => feed.resolve_converted(symbol, conversion, reply),
842        None => feed.resolve(symbol, reply),
843    }
844}
845
846/// How long to wait before retrying a request that has failed `failures` times: doubling from a
847/// second up to a minute, and no sooner than the feed asked.
848pub(super) fn retry_delay(failures: u32, error: &FeedError) -> Duration {
849    let backoff = Duration::from_secs(1 << failures.min(6)).min(Duration::from_secs(60));
850    match error {
851        FeedError::RateLimited {
852            retry_after: Some(after),
853        } => backoff.max(*after),
854        _ => backoff,
855    }
856}
857
858/// `wake`, also after a panic while it was held.
859fn lock(wake: &Mutex<Option<ViewportId>>) -> std::sync::MutexGuard<'_, Option<ViewportId>> {
860    wake.lock().unwrap_or_else(PoisonError::into_inner)
861}
862
863#[cfg(test)]
864mod tests;
egui_charts/widget/feed/mod.rs

egui_charts/widget/feed/
mod.rs
