navi_vm/
data_provider.rs

1//! [`DataProvider`] trait and associated types.
2//!
3//! Implement [`DataProvider`] and pass it as the first argument to
4//! [`Instance::builder`](crate::Instance::builder) to supply candlestick data
5//! for both the main chart and `request.security()` calls.
6
7use std::{fmt, rc::Rc};
8
9use async_trait::async_trait;
10use futures_util::{Stream, StreamExt, stream::LocalBoxStream};
11use navi_types::{
12    AuxDataItem, Candlestick, DataArgs, DividendsField, EarningsField, SplitsField, SymbolInfo,
13    SymbolType, SyminfoFields, TickItem, TimeUnit, VolumeType,
14};
15use serde::{Deserialize, Serialize};
16use time::OffsetDateTime;
17
18use crate::{CandlestickItem, Currency, Market, TimeFrame, market_ext::MarketExt};
19
20/// Partial symbol metadata returned by [`DataProvider::symbol_info`].
21///
22/// All fields are `Option`. Any field left as `None` will fall back to the
23/// default value derived from the symbol string (exchange prefix → market →
24/// `market.default_*()` helpers).
25#[derive(Debug, Clone, Default, Serialize, Deserialize)]
26#[serde(rename_all = "camelCase", default)]
27pub struct PartialSymbolInfo {
28    /// The market/exchange the symbol belongs to.
29    pub market: Option<Market>,
30    /// Human-readable description of the symbol (e.g. `"Apple Inc."`).
31    pub description: Option<String>,
32    /// The type of instrument (stock, futures, crypto, …).
33    pub type_: Option<SymbolType>,
34    /// ISO 3166-1 alpha-2 country code of the exchange (e.g. `"US"`, `"HK"`).
35    pub country: Option<String>,
36    /// International Securities Identification Number.
37    pub isin: Option<String>,
38    /// Root identifier for derivative instruments (e.g. `"ES"` for `"ESZ4"`).
39    pub root: Option<String>,
40    /// Numerator of the `syminfo.mintick` formula (`min_move / price_scale`).
41    pub min_move: Option<i32>,
42    /// Denominator of the `syminfo.mintick` formula.
43    pub price_scale: Option<i32>,
44    /// Point value multiplier (usually `1.0`; relevant for futures).
45    pub point_value: Option<f64>,
46    /// Currency of the symbol's prices.
47    pub currency: Option<Currency>,
48    /// Expiration date of the current futures contract.
49    pub expiration_date: Option<OffsetDateTime>,
50    /// Ticker identifier of the underlying contract.
51    pub current_contract: Option<String>,
52    /// Base currency for forex/crypto pairs.
53    pub base_currency: Option<Currency>,
54    /// Number of employees (stocks only).
55    pub employees: Option<i64>,
56    /// Industry classification.
57    pub industry: Option<String>,
58    /// Sector classification.
59    pub sector: Option<String>,
60    /// Minimum contract size.
61    pub min_contract: Option<f64>,
62    /// Volume type interpretation.
63    pub volume_type: Option<VolumeType>,
64    /// Number of shareholders.
65    pub shareholders: Option<i64>,
66    /// Free-float shares outstanding.
67    pub shares_outstanding_float: Option<f64>,
68    /// Total shares outstanding.
69    pub shares_outstanding_total: Option<f64>,
70    /// Number of analyst buy recommendations.
71    pub recommendations_buy: Option<i32>,
72    /// Number of analyst strong-buy recommendations.
73    pub recommendations_buy_strong: Option<i32>,
74    /// Number of analyst hold recommendations.
75    pub recommendations_hold: Option<i32>,
76    /// Number of analyst sell recommendations.
77    pub recommendations_sell: Option<i32>,
78    /// Number of analyst strong-sell recommendations.
79    pub recommendations_sell_strong: Option<i32>,
80    /// Date of the latest analyst recommendations update.
81    pub recommendations_date: Option<OffsetDateTime>,
82    /// Total number of analyst recommendations.
83    pub recommendations_total: Option<i32>,
84    /// Average analyst price target.
85    pub target_price_average: Option<f64>,
86    /// Date of the latest price target update.
87    pub target_price_date: Option<OffsetDateTime>,
88    /// Total number of price target estimates.
89    pub target_price_estimates: Option<f64>,
90    /// Highest analyst price target.
91    pub target_price_high: Option<f64>,
92    /// Lowest analyst price target.
93    pub target_price_low: Option<f64>,
94    /// Median analyst price target.
95    pub target_price_median: Option<f64>,
96}
97
98impl PartialSymbolInfo {
99    /// Merges this partial metadata with a symbol string to produce a full
100    /// [`SymbolInfo`].
101    ///
102    /// Any `None` field falls back to the default value derived from the
103    /// symbol's exchange prefix via the market helpers.
104    pub(crate) fn into_symbol_info(self, symbol: &str) -> Result<SymbolInfo, navi_types::Error> {
105        use navi_types::symbol::Symbol;
106        let symbol: Symbol = symbol
107            .parse()
108            .map_err(|_| navi_types::Error::InvalidSymbol)?;
109        let market = match self.market {
110            Some(m) => m,
111            None => symbol
112                .prefix()
113                .parse()
114                .map_err(|_| navi_types::Error::UnknownMarket)?,
115        };
116        Ok(SymbolInfo {
117            symbol,
118            market,
119            description: self.description,
120            type_: self.type_.or_else(|| Some(market.default_type())),
121            country: self
122                .country
123                .or_else(|| Some(market.default_country().to_string())),
124            isin: self.isin,
125            root: self.root,
126            min_move: self.min_move.unwrap_or_else(|| market.default_min_move()),
127            price_scale: self
128                .price_scale
129                .unwrap_or_else(|| market.default_price_scale()),
130            point_value: self.point_value.unwrap_or(1.0),
131            currency: self.currency.unwrap_or_else(|| market.default_currency()),
132            expiration_date: self.expiration_date,
133            current_contract: self.current_contract,
134            base_currency: self.base_currency,
135            employees: self.employees,
136            industry: self.industry,
137            sector: self.sector,
138            min_contract: self.min_contract,
139            recommendations_buy: self.recommendations_buy,
140            recommendations_buy_strong: self.recommendations_buy_strong,
141            recommendations_hold: self.recommendations_hold,
142            recommendations_sell: self.recommendations_sell,
143            recommendations_sell_strong: self.recommendations_sell_strong,
144            recommendations_date: self.recommendations_date,
145            recommendations_total: self.recommendations_total,
146            shareholders: self.shareholders,
147            shares_outstanding_float: self.shares_outstanding_float,
148            shares_outstanding_total: self.shares_outstanding_total,
149            target_price_average: self.target_price_average,
150            target_price_date: self.target_price_date,
151            target_price_estimates: self.target_price_estimates,
152            target_price_high: self.target_price_high,
153            target_price_low: self.target_price_low,
154            target_price_median: self.target_price_median,
155            volume_type: self
156                .volume_type
157                .or_else(|| Some(market.default_volume_type())),
158        })
159    }
160}
161
162/// Error returned by [`DataProvider`] methods.
163///
164/// Use `std::convert::Infallible` as `E` when the provider can only fail with
165/// [`InvalidSymbol`](DataProviderError::InvalidSymbol) and never with
166/// [`Other`](DataProviderError::Other).
167#[non_exhaustive]
168#[derive(Debug, thiserror::Error)]
169pub enum DataProviderError<E>
170where
171    E: fmt::Display + fmt::Debug,
172{
173    /// The requested symbol does not exist or is not supported.
174    ///
175    /// When `request.security()` is called with `ignore_invalid_symbol = true`,
176    /// this causes the function to return `na` instead of a runtime error.
177    #[error("invalid symbol: {0}")]
178    InvalidSymbol(String),
179    /// Any other provider-side error.
180    #[error("{0}")]
181    Other(E),
182}
183
184/// Supplies candlestick data and symbol metadata to the VM.
185///
186/// Implement this trait and pass it as the first argument to
187/// [`Instance::builder`](crate::Instance::builder). It serves two purposes:
188/// main chart K-lines consumed by [`Instance::run`](crate::Instance::run) and
189/// MTF/cross-symbol data for `request.security()` calls.
190///
191/// For simple single-symbol use cases, pass a `Vec<Candlestick>` directly —
192/// it implements `DataProvider` and wraps bars as
193/// [`CandlestickItem::Bar`].
194///
195/// # Contract
196///
197/// - [`candlesticks`](DataProvider::candlesticks) must return bars in **strict
198///   ascending time order**.
199/// - The stream should start at or before `from_time` so that MTF bar state is
200///   initialised even before the first chart bar.
201/// - Returning `Err(DataProviderError::InvalidSymbol(_))` causes
202///   `request.security()` to return `na` when `ignore_invalid_symbol = true`,
203///   or a runtime error otherwise.
204#[async_trait(?Send)]
205#[allow(clippy::type_complexity)]
206pub trait DataProvider {
207    /// The error payload type for [`DataProviderError::Other`].
208    type Error: fmt::Display + fmt::Debug + 'static;
209
210    /// Returns partial symbol metadata for the given symbol string.
211    ///
212    /// `fields` is the set of `syminfo` property fields that the compiled
213    /// script actually accesses.  Implementations **may** use this to fetch
214    /// only the required data, but are free to always return all fields.
215    ///
216    /// Any `None` field falls back to a default derived from the symbol's
217    /// exchange prefix. Return `PartialSymbolInfo::default()` if you only want
218    /// the defaults.
219    async fn symbol_info(
220        &self,
221        symbol: String,
222        fields: SyminfoFields,
223    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>>;
224
225    /// Returns a stream of candlestick items for the given symbol and
226    /// timeframe.
227    ///
228    /// - `symbol`    — e.g. `"BINANCE:BTCUSDT"`, `"NASDAQ:AAPL"`
229    /// - `timeframe` — the requested timeframe
230    /// - `from_time` — epoch-millisecond timestamp of the first chart bar;
231    ///   start at or before this time
232    /// - `count`     — optional cap for how many recent historical bars to load
233    ///   for a shared request stream; `None` means no provider-side cap
234    ///
235    /// Items must be yielded in **strict ascending time order**.
236    ///
237    /// Returns `Err(DataProviderError::InvalidSymbol(_))` at creation time
238    /// when the symbol is not recognised. This is checked before any stream
239    /// items are yielded.
240    ///
241    /// # Finite vs. infinite streams
242    ///
243    /// The stream may be **finite** (all items are yielded and the stream
244    /// ends) or **infinite** (the stream never terminates and keeps
245    /// producing `Realtime` items as new ticks arrive).
246    ///
247    /// - **Finite** — typical for historical backtests. The stream yields
248    ///   `Confirmed` bars, then `HistoryEnd`, then terminates. If a `Realtime`
249    ///   bar is still open when the stream ends, the VM confirms it
250    ///   automatically. [`Instance::run`](crate::Instance::run) returns when
251    ///   the stream is exhausted.
252    /// - **Infinite** — typical for live trading / real-time monitoring. The
253    ///   stream yields historical bars followed by `HistoryEnd`, then keeps
254    ///   producing `Realtime` items indefinitely as new market ticks arrive.
255    ///   [`Instance::run`](crate::Instance::run) runs until the stream is
256    ///   dropped or the caller stops polling.
257    ///
258    /// # Stream protocol
259    ///
260    /// The expected sequence is:
261    ///
262    /// ```text
263    /// Bar(t0)  Bar(t1)  …  Bar(tN-1)  Bar(tN)   <- bars sent before HistoryEnd
264    /// HistoryEnd                                  <- boundary marker (once)
265    /// Bar(tN)                         <- realtime update for the forming bar
266    ///                                    (tN may equal the last historical ts)
267    /// Bar(tA)  (tA > tN)              <- new period: VM confirms tN, opens tA
268    /// Bar(tA)                         <- in-place update for tA
269    /// …                               <- finite: stream ends
270    ///                                    infinite: keeps producing
271    /// ```
272    ///
273    /// Key rules:
274    ///
275    /// - **[`Bar`](crate::CandlestickItem::Bar)** — the role depends on
276    ///   position relative to `HistoryEnd`. Before `HistoryEnd`, bars `t0`
277    ///   through `tN-1` are fully closed historical bars. The last bar `tN`
278    ///   (the one immediately before `HistoryEnd`) may be a currently forming
279    ///   bar — the VM executes it in realtime mode for this reason. After
280    ///   `HistoryEnd`, `Bar` items represent the currently forming bar.
281    ///   Multiple consecutive `Bar` items with the *same* timestamp are treated
282    ///   as in-place updates. A `Bar` with a *later* timestamp implicitly
283    ///   confirms the previous bar and starts a new one.
284    /// - **[`HistoryEnd`](crate::CandlestickItem::HistoryEnd)** — emitted once,
285    ///   signals that the server has sent all currently available bars. It does
286    ///   **not** mean the last bar is a closed period — if the current period
287    ///   is still forming, the last bar before `HistoryEnd` will be that
288    ///   forming bar, and realtime updates for it will follow with the same
289    ///   timestamp. If the stream ends after `HistoryEnd` with an open realtime
290    ///   bar, the VM confirms it automatically.
291    fn candlesticks(
292        &self,
293        symbol: String,
294        timeframe: TimeFrame,
295        from_time: i64,
296        count: Option<usize>,
297    ) -> Result<
298        impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
299        DataProviderError<Self::Error>,
300    >;
301
302    /// Returns a stream of tick events for the given symbol.
303    ///
304    /// - `symbol`    — e.g. `"BINANCE:BTCUSDT"`, `"NASDAQ:AAPL"`
305    /// - `from_time` — epoch-millisecond timestamp of the first chart bar;
306    ///   start at or before this time
307    /// - `count`     — optional cap for how many recent historical ticks to
308    ///   load; `None` means no provider-side cap
309    ///
310    /// The stream protocol is identical to
311    /// [`candlesticks`](Self::candlesticks): historical [`TickItem::Tick`]
312    /// events are followed by exactly one [`TickItem::HistoryEnd`] marker,
313    /// then optional realtime ticks.
314    ///
315    /// This method is only called by the VM when a tick-based timeframe (`T`
316    /// suffix) is requested. The default implementation returns an empty stream
317    /// (no tick data available).
318    fn ticks(
319        &self,
320        _symbol: String,
321        _from_time: i64,
322        _count: Option<usize>,
323    ) -> Result<
324        impl Stream<Item = Result<TickItem, DataProviderError<Self::Error>>> + 'static,
325        DataProviderError<Self::Error>,
326    > {
327        Ok(futures_util::stream::empty())
328    }
329
330    /// Returns a stream of historical exchange rate data points for
331    /// converting from `from` currency to `to` currency.
332    ///
333    /// Default: empty stream (no data available).
334    fn currency_rate(
335        &self,
336        _from: Currency,
337        _to: Currency,
338        _from_time: i64,
339    ) -> Result<
340        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
341        DataProviderError<Self::Error>,
342    > {
343        Ok(futures_util::stream::empty())
344    }
345
346    /// Returns a stream of financial report data points (e.g. quarterly
347    /// revenue, EPS) for the given symbol and financial identifier.
348    ///
349    /// Default: empty stream (no data available).
350    fn financial(
351        &self,
352        _symbol: String,
353        _financial_id: String,
354        _period: String,
355        _currency: Option<Currency>,
356        _from_time: i64,
357    ) -> Result<
358        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
359        DataProviderError<Self::Error>,
360    > {
361        Ok(futures_util::stream::empty())
362    }
363
364    /// Returns a stream of earnings data points for the given symbol.
365    ///
366    /// Default: empty stream (no data available).
367    fn earnings(
368        &self,
369        _symbol: String,
370        _field: EarningsField,
371        _currency: Option<Currency>,
372        _from_time: i64,
373    ) -> Result<
374        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
375        DataProviderError<Self::Error>,
376    > {
377        Ok(futures_util::stream::empty())
378    }
379
380    /// Returns a stream of dividend data points for the given symbol.
381    ///
382    /// Default: empty stream (no data available).
383    fn dividends(
384        &self,
385        _symbol: String,
386        _field: DividendsField,
387        _currency: Option<Currency>,
388        _from_time: i64,
389    ) -> Result<
390        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
391        DataProviderError<Self::Error>,
392    > {
393        Ok(futures_util::stream::empty())
394    }
395
396    /// Returns a stream of macroeconomic data points for the given country
397    /// and field.
398    ///
399    /// Default: empty stream (no data available).
400    fn economic(
401        &self,
402        _country_code: String,
403        _field: String,
404        _from_time: i64,
405    ) -> Result<
406        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
407        DataProviderError<Self::Error>,
408    > {
409        Ok(futures_util::stream::empty())
410    }
411
412    /// Returns a stream of stock split data points for the given symbol.
413    ///
414    /// Default: empty stream (no data available).
415    fn splits(
416        &self,
417        _symbol: String,
418        _field: SplitsField,
419        _from_time: i64,
420    ) -> Result<
421        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
422        DataProviderError<Self::Error>,
423    > {
424        Ok(futures_util::stream::empty())
425    }
426
427    /// Returns a custom data stream identified by `function` and `args`.
428    ///
429    /// Called by `request.data(function, args)` in Navi.  Implement
430    /// this method to expose arbitrary time-series data to scripts.
431    ///
432    /// - `function` — the function name from the `request.data` call site.
433    /// - `args`     — typed key/value parameters from the `map<string, any>`
434    ///   argument.  Use [`DataArgs::deserialize`] to convert them into a typed
435    ///   struct.
436    /// - `from_time` — epoch-millisecond timestamp of the first chart bar.
437    ///
438    /// Default: empty stream (no data available).
439    fn data(
440        &self,
441        _function: String,
442        _args: DataArgs,
443        _from_time: i64,
444    ) -> Result<
445        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
446        DataProviderError<Self::Error>,
447    > {
448        Ok(futures_util::stream::empty())
449    }
450}
451
452// ── HistoryProvider ────────────────────────────────────────────────────────
453
454/// Supplies older historical bars for incremental history extension.
455///
456/// Implement alongside [`DataProvider`] to enable efficient `extend_history`:
457/// only the bars older than the current cache are fetched, then merged with
458/// cached bars before scripts are re-executed.
459#[async_trait(?Send)]
460pub trait HistoryProvider {
461    /// The error payload type; should match [`DataProvider::Error`] on the
462    /// same implementing type.
463    type Error: fmt::Display + fmt::Debug + 'static;
464
465    /// Fetch up to `count` bars for `symbol`/`timeframe` where
466    /// `bar.time < before_time`, returned in **strict ascending time order**
467    /// (oldest first).
468    ///
469    /// Returning an empty `Vec` signals that no older data is available
470    /// (history exhausted).
471    async fn history_bars_before(
472        &self,
473        symbol: String,
474        timeframe: TimeFrame,
475        before_time: i64,
476        count: usize,
477    ) -> Result<Vec<Candlestick>, DataProviderError<Self::Error>>;
478}
479
480// ── HistoryProvider impls for built-in types ───────────────────────────────
481
482#[async_trait(?Send)]
483impl HistoryProvider for Vec<Candlestick> {
484    type Error = std::convert::Infallible;
485
486    async fn history_bars_before(
487        &self,
488        _symbol: String,
489        _timeframe: TimeFrame,
490        before_time: i64,
491        count: usize,
492    ) -> Result<Vec<Candlestick>, DataProviderError<Self::Error>> {
493        let end = self.partition_point(|c| c.time < before_time);
494        let start = end.saturating_sub(count);
495        Ok(self[start..end].to_vec())
496    }
497}
498
499#[async_trait(?Send)]
500impl HistoryProvider for Vec<CandlestickItem> {
501    type Error = std::convert::Infallible;
502
503    async fn history_bars_before(
504        &self,
505        _symbol: String,
506        _timeframe: TimeFrame,
507        before_time: i64,
508        count: usize,
509    ) -> Result<Vec<Candlestick>, DataProviderError<Self::Error>> {
510        let bars: Vec<Candlestick> = self
511            .iter()
512            .filter_map(|item| {
513                if let CandlestickItem::Bar(c) = item {
514                    if c.time < before_time { Some(*c) } else { None }
515                } else {
516                    None
517                }
518            })
519            .collect();
520        let start = bars.len().saturating_sub(count);
521        Ok(bars[start..].to_vec())
522    }
523}
524
525#[async_trait(?Send)]
526impl<P> HistoryProvider for Rc<P>
527where
528    P: HistoryProvider + 'static,
529{
530    type Error = P::Error;
531
532    async fn history_bars_before(
533        &self,
534        symbol: String,
535        timeframe: TimeFrame,
536        before_time: i64,
537        count: usize,
538    ) -> Result<Vec<Candlestick>, DataProviderError<Self::Error>> {
539        (**self)
540            .history_bars_before(symbol, timeframe, before_time, count)
541            .await
542    }
543}
544
545/// Internal error produced by [`InternalProvider`].
546///
547/// Unlike [`DataProviderError`], this always carries full context. It is
548/// produced by the blanket [`InternalProvider`] impl.
549#[derive(Debug, thiserror::Error)]
550pub(crate) enum InternalProviderError {
551    /// The symbol does not exist or is not supported.
552    #[error("invalid symbol: {0}")]
553    InvalidSymbol(String),
554    /// `get_symbol_info` failed with a non-InvalidSymbol error.
555    #[error("error fetching symbol info for {symbol}: {message}")]
556    SymbolInfoError { symbol: String, message: String },
557    /// `get_candlesticks` failed with a non-InvalidSymbol error.
558    #[error("error fetching {symbol}/{timeframe}: {message}")]
559    CandlesticksError {
560        symbol: String,
561        timeframe: TimeFrame,
562        message: String,
563    },
564    /// An auxiliary data stream failed with a non-InvalidSymbol error.
565    #[error("auxiliary data error: {message}")]
566    AuxDataError { message: String },
567}
568
569/// Internal trait used to store `DataProvider` as a `Box<dyn
570/// InternalProvider>`.
571///
572/// Automatically implemented for all `T: DataProvider + 'static`. Bridges the
573/// public `DataProvider` API (which uses `DataProviderError<E>`) to the
574/// internal `InternalProviderError` used by the VM.
575#[async_trait(?Send)]
576pub(crate) trait InternalProvider {
577    async fn symbol_info(
578        &self,
579        _symbol: String,
580        _fields: SyminfoFields,
581    ) -> Result<PartialSymbolInfo, InternalProviderError> {
582        Ok(PartialSymbolInfo::default())
583    }
584
585    fn candlesticks(
586        &self,
587        _symbol: &str,
588        _timeframe: TimeFrame,
589        _from_time: i64,
590        _count: Option<usize>,
591    ) -> Result<
592        LocalBoxStream<'static, Result<CandlestickItem, InternalProviderError>>,
593        InternalProviderError,
594    > {
595        Ok(Box::pin(futures_util::stream::empty()))
596    }
597
598    fn currency_rate(
599        &self,
600        _from: Currency,
601        _to: Currency,
602        _from_time: i64,
603    ) -> Result<
604        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
605        InternalProviderError,
606    > {
607        Ok(Box::pin(futures_util::stream::empty()))
608    }
609
610    fn financial(
611        &self,
612        _symbol: String,
613        _financial_id: String,
614        _period: String,
615        _currency: Option<Currency>,
616        _from_time: i64,
617    ) -> Result<
618        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
619        InternalProviderError,
620    > {
621        Ok(Box::pin(futures_util::stream::empty()))
622    }
623
624    fn earnings(
625        &self,
626        _symbol: String,
627        _field: EarningsField,
628        _currency: Option<Currency>,
629        _from_time: i64,
630    ) -> Result<
631        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
632        InternalProviderError,
633    > {
634        Ok(Box::pin(futures_util::stream::empty()))
635    }
636
637    fn dividends(
638        &self,
639        _symbol: String,
640        _field: DividendsField,
641        _currency: Option<Currency>,
642        _from_time: i64,
643    ) -> Result<
644        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
645        InternalProviderError,
646    > {
647        Ok(Box::pin(futures_util::stream::empty()))
648    }
649
650    fn economic(
651        &self,
652        _country_code: String,
653        _field: String,
654        _from_time: i64,
655    ) -> Result<
656        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
657        InternalProviderError,
658    > {
659        Ok(Box::pin(futures_util::stream::empty()))
660    }
661
662    fn splits(
663        &self,
664        _symbol: String,
665        _field: SplitsField,
666        _from_time: i64,
667    ) -> Result<
668        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
669        InternalProviderError,
670    > {
671        Ok(Box::pin(futures_util::stream::empty()))
672    }
673
674    fn data(
675        &self,
676        _function: String,
677        _args: DataArgs,
678        _from_time: i64,
679    ) -> Result<
680        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
681        InternalProviderError,
682    > {
683        Ok(Box::pin(futures_util::stream::empty()))
684    }
685}
686
687/// Maps a `DataProviderError<E>` from an aux data stream item into an
688/// `InternalProviderError`.
689fn map_aux_stream_error<E>(
690    result: Result<AuxDataItem, DataProviderError<E>>,
691) -> Result<AuxDataItem, InternalProviderError>
692where
693    E: fmt::Display + fmt::Debug,
694{
695    result.map_err(|e| match e {
696        DataProviderError::InvalidSymbol(msg) => InternalProviderError::InvalidSymbol(msg),
697        DataProviderError::Other(e) => InternalProviderError::AuxDataError {
698            message: e.to_string(),
699        },
700    })
701}
702
703/// Maps a `DataProviderError<E>` from aux stream creation into an
704/// `InternalProviderError`.
705fn map_aux_creation_error<E>(e: DataProviderError<E>) -> InternalProviderError
706where
707    E: fmt::Display + fmt::Debug,
708{
709    match e {
710        DataProviderError::InvalidSymbol(msg) => InternalProviderError::InvalidSymbol(msg),
711        DataProviderError::Other(e) => InternalProviderError::AuxDataError {
712            message: e.to_string(),
713        },
714    }
715}
716
717#[async_trait(?Send)]
718impl<T> InternalProvider for T
719where
720    T: DataProvider + 'static,
721{
722    async fn symbol_info(
723        &self,
724        symbol: String,
725        fields: SyminfoFields,
726    ) -> Result<PartialSymbolInfo, InternalProviderError> {
727        DataProvider::symbol_info(self, symbol.clone(), fields)
728            .await
729            .map_err(|e| match e {
730                DataProviderError::InvalidSymbol(msg) => InternalProviderError::InvalidSymbol(msg),
731                DataProviderError::Other(e) => InternalProviderError::SymbolInfoError {
732                    symbol,
733                    message: e.to_string(),
734                },
735            })
736    }
737
738    fn candlesticks(
739        &self,
740        symbol: &str,
741        timeframe: TimeFrame,
742        from_time: i64,
743        count: Option<usize>,
744    ) -> Result<
745        LocalBoxStream<'static, Result<CandlestickItem, InternalProviderError>>,
746        InternalProviderError,
747    > {
748        let symbol_owned = symbol.to_string();
749
750        // Route tick-based timeframes through the ticks() stream.
751        if timeframe.unit == TimeUnit::Tick {
752            let n = timeframe.quantity.get() as usize;
753            let raw =
754                DataProvider::ticks(self, symbol_owned.clone(), from_time, count).map_err(|e| {
755                    match e {
756                        DataProviderError::InvalidSymbol(msg) => {
757                            InternalProviderError::InvalidSymbol(msg)
758                        }
759                        DataProviderError::Other(e) => InternalProviderError::AuxDataError {
760                            message: e.to_string(),
761                        },
762                    }
763                })?;
764            let mapped = raw.map(move |result| {
765                result.map_err(|e| match e {
766                    DataProviderError::InvalidSymbol(msg) => {
767                        InternalProviderError::InvalidSymbol(msg)
768                    }
769                    DataProviderError::Other(e) => InternalProviderError::AuxDataError {
770                        message: e.to_string(),
771                    },
772                })
773            });
774            return Ok(Box::pin(synthesize_ticks(mapped, n)));
775        }
776
777        let stream =
778            DataProvider::candlesticks(self, symbol_owned.clone(), timeframe, from_time, count)
779                .map_err(|e| match e {
780                    DataProviderError::InvalidSymbol(msg) => {
781                        InternalProviderError::InvalidSymbol(msg)
782                    }
783                    DataProviderError::Other(e) => InternalProviderError::CandlesticksError {
784                        symbol: symbol_owned.clone(),
785                        timeframe,
786                        message: e.to_string(),
787                    },
788                })?;
789
790        Ok(Box::pin(stream.map(move |result| {
791            result.map_err(|e| match e {
792                DataProviderError::InvalidSymbol(msg) => InternalProviderError::InvalidSymbol(msg),
793                DataProviderError::Other(e) => InternalProviderError::CandlesticksError {
794                    symbol: symbol_owned.clone(),
795                    timeframe,
796                    message: e.to_string(),
797                },
798            })
799        })))
800    }
801
802    fn currency_rate(
803        &self,
804        from: Currency,
805        to: Currency,
806        from_time: i64,
807    ) -> Result<
808        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
809        InternalProviderError,
810    > {
811        let stream = DataProvider::currency_rate(self, from, to, from_time)
812            .map_err(map_aux_creation_error)?;
813        Ok(Box::pin(stream.map(map_aux_stream_error)))
814    }
815
816    fn financial(
817        &self,
818        symbol: String,
819        financial_id: String,
820        period: String,
821        currency: Option<Currency>,
822        from_time: i64,
823    ) -> Result<
824        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
825        InternalProviderError,
826    > {
827        let stream =
828            DataProvider::financial(self, symbol, financial_id, period, currency, from_time)
829                .map_err(map_aux_creation_error)?;
830        Ok(Box::pin(stream.map(map_aux_stream_error)))
831    }
832
833    fn earnings(
834        &self,
835        symbol: String,
836        field: EarningsField,
837        currency: Option<Currency>,
838        from_time: i64,
839    ) -> Result<
840        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
841        InternalProviderError,
842    > {
843        let stream = DataProvider::earnings(self, symbol, field, currency, from_time)
844            .map_err(map_aux_creation_error)?;
845        Ok(Box::pin(stream.map(map_aux_stream_error)))
846    }
847
848    fn dividends(
849        &self,
850        symbol: String,
851        field: DividendsField,
852        currency: Option<Currency>,
853        from_time: i64,
854    ) -> Result<
855        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
856        InternalProviderError,
857    > {
858        let stream = DataProvider::dividends(self, symbol, field, currency, from_time)
859            .map_err(map_aux_creation_error)?;
860        Ok(Box::pin(stream.map(map_aux_stream_error)))
861    }
862
863    fn economic(
864        &self,
865        country_code: String,
866        field: String,
867        from_time: i64,
868    ) -> Result<
869        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
870        InternalProviderError,
871    > {
872        let stream = DataProvider::economic(self, country_code, field, from_time)
873            .map_err(map_aux_creation_error)?;
874        Ok(Box::pin(stream.map(map_aux_stream_error)))
875    }
876
877    fn splits(
878        &self,
879        symbol: String,
880        field: SplitsField,
881        from_time: i64,
882    ) -> Result<
883        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
884        InternalProviderError,
885    > {
886        let stream =
887            DataProvider::splits(self, symbol, field, from_time).map_err(map_aux_creation_error)?;
888        Ok(Box::pin(stream.map(map_aux_stream_error)))
889    }
890
891    fn data(
892        &self,
893        function: String,
894        args: DataArgs,
895        from_time: i64,
896    ) -> Result<
897        LocalBoxStream<'static, Result<AuxDataItem, InternalProviderError>>,
898        InternalProviderError,
899    > {
900        let stream =
901            DataProvider::data(self, function, args, from_time).map_err(map_aux_creation_error)?;
902        Ok(Box::pin(stream.map(map_aux_stream_error)))
903    }
904}
905
906/// `Vec<CandlestickItem>` implements [`DataProvider`] by yielding the items
907/// as-is on every `candlesticks()` call, regardless of symbol/timeframe.
908///
909/// Use this when you need to mix [`CandlestickItem::Bar`] and
910/// [`CandlestickItem::HistoryEnd`] in a single stream.
911/// For plain historical bars, prefer `Vec<Candlestick>`.
912#[async_trait(?Send)]
913impl DataProvider for Vec<CandlestickItem> {
914    type Error = std::convert::Infallible;
915
916    async fn symbol_info(
917        &self,
918        _symbol: String,
919        _fields: SyminfoFields,
920    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
921        Ok(PartialSymbolInfo::default())
922    }
923
924    fn candlesticks(
925        &self,
926        _symbol: String,
927        _timeframe: TimeFrame,
928        _from_time: i64,
929        count: Option<usize>,
930    ) -> Result<
931        impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
932        DataProviderError<Self::Error>,
933    > {
934        let mut items = self.clone();
935        if let Some(count) = count {
936            let history_end_idx = items
937                .iter()
938                .position(|item| matches!(item, CandlestickItem::HistoryEnd))
939                .unwrap_or(items.len());
940            let bar_indices: Vec<usize> = items[..history_end_idx]
941                .iter()
942                .enumerate()
943                .filter_map(|(idx, item)| matches!(item, CandlestickItem::Bar(_)).then_some(idx))
944                .collect();
945            if bar_indices.len() > count {
946                let first_keep = bar_indices[bar_indices.len() - count];
947                items.drain(..first_keep);
948            }
949        }
950        Ok(futures_util::stream::iter(items.into_iter().map(Ok)))
951    }
952}
953
954/// `Rc<P>` implements [`DataProvider`] by delegating to the inner `P`.
955///
956/// Allows sharing a single provider instance between multiple
957/// [`Instance`](crate::Instance) builders or background tasks without cloning
958/// the provider value itself.
959#[async_trait(?Send)]
960impl<P> DataProvider for Rc<P>
961where
962    P: DataProvider + 'static,
963{
964    type Error = P::Error;
965
966    async fn symbol_info(
967        &self,
968        symbol: String,
969        fields: SyminfoFields,
970    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
971        (**self).symbol_info(symbol, fields).await
972    }
973
974    fn candlesticks(
975        &self,
976        symbol: String,
977        timeframe: TimeFrame,
978        from_time: i64,
979        count: Option<usize>,
980    ) -> Result<
981        impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
982        DataProviderError<Self::Error>,
983    > {
984        (**self).candlesticks(symbol, timeframe, from_time, count)
985    }
986
987    fn ticks(
988        &self,
989        symbol: String,
990        from_time: i64,
991        count: Option<usize>,
992    ) -> Result<
993        impl Stream<Item = Result<TickItem, DataProviderError<Self::Error>>> + 'static,
994        DataProviderError<Self::Error>,
995    > {
996        (**self).ticks(symbol, from_time, count)
997    }
998
999    fn currency_rate(
1000        &self,
1001        from: Currency,
1002        to: Currency,
1003        from_time: i64,
1004    ) -> Result<
1005        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
1006        DataProviderError<Self::Error>,
1007    > {
1008        (**self).currency_rate(from, to, from_time)
1009    }
1010
1011    fn financial(
1012        &self,
1013        symbol: String,
1014        financial_id: String,
1015        period: String,
1016        currency: Option<Currency>,
1017        from_time: i64,
1018    ) -> Result<
1019        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
1020        DataProviderError<Self::Error>,
1021    > {
1022        (**self).financial(symbol, financial_id, period, currency, from_time)
1023    }
1024
1025    fn earnings(
1026        &self,
1027        symbol: String,
1028        field: EarningsField,
1029        currency: Option<Currency>,
1030        from_time: i64,
1031    ) -> Result<
1032        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
1033        DataProviderError<Self::Error>,
1034    > {
1035        (**self).earnings(symbol, field, currency, from_time)
1036    }
1037
1038    fn dividends(
1039        &self,
1040        symbol: String,
1041        field: DividendsField,
1042        currency: Option<Currency>,
1043        from_time: i64,
1044    ) -> Result<
1045        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
1046        DataProviderError<Self::Error>,
1047    > {
1048        (**self).dividends(symbol, field, currency, from_time)
1049    }
1050
1051    fn economic(
1052        &self,
1053        country_code: String,
1054        field: String,
1055        from_time: i64,
1056    ) -> Result<
1057        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
1058        DataProviderError<Self::Error>,
1059    > {
1060        (**self).economic(country_code, field, from_time)
1061    }
1062
1063    fn splits(
1064        &self,
1065        symbol: String,
1066        field: SplitsField,
1067        from_time: i64,
1068    ) -> Result<
1069        impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
1070        DataProviderError<Self::Error>,
1071    > {
1072        (**self).splits(symbol, field, from_time)
1073    }
1074}
1075
1076/// `Vec<Candlestick>` implements [`DataProvider`] for the simple case of
1077/// running a script against a fixed set of historical bars.
1078///
1079/// All `candlesticks()` requests (regardless of symbol/timeframe) receive the
1080/// same bars wrapped as [`CandlestickItem::Bar`], followed by a
1081/// [`CandlestickItem::HistoryEnd`] marker.
1082///
1083/// For `script_info()` queries that need no data, pass `vec![]`.
1084#[async_trait(?Send)]
1085impl DataProvider for Vec<Candlestick> {
1086    type Error = std::convert::Infallible;
1087
1088    async fn symbol_info(
1089        &self,
1090        _symbol: String,
1091        _fields: SyminfoFields,
1092    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
1093        Ok(PartialSymbolInfo::default())
1094    }
1095
1096    fn candlesticks(
1097        &self,
1098        _symbol: String,
1099        _timeframe: TimeFrame,
1100        _from_time: i64,
1101        count: Option<usize>,
1102    ) -> Result<
1103        impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
1104        DataProviderError<Self::Error>,
1105    > {
1106        let bars = if let Some(count) = count {
1107            let start = self.len().saturating_sub(count);
1108            self[start..].to_vec()
1109        } else {
1110            self.clone()
1111        };
1112        let bar_items = bars.into_iter().map(|c| Ok(CandlestickItem::Bar(c)));
1113        let end = std::iter::once(Ok(CandlestickItem::HistoryEnd));
1114        Ok(futures_util::stream::iter(bar_items.chain(end)))
1115    }
1116}
1117
1118/// Stream adapter that converts a tick stream into a [`CandlestickItem`]
1119/// stream.
1120///
1121/// - **1T (`n == 1`)**: each tick becomes one `Bar`; `HistoryEnd` is forwarded.
1122/// - **nT (`n > 1`)**: ticks are accumulated into groups of `n`. During
1123///   history, each complete group of `n` ticks emits one `Bar`. After
1124///   `HistoryEnd`, every incoming tick emits an updated `Bar` with the current
1125///   accumulated OHLCV; the accumulator resets after `n` ticks are complete.
1126#[pin_project::pin_project]
1127struct SynthesizeTicksStream<S> {
1128    #[pin]
1129    raw: S,
1130    /// Number of ticks per bar (1 for 1T).
1131    n: usize,
1132    /// `true` after the inner stream has emitted `TickItem::HistoryEnd`.
1133    history_ended: bool,
1134    /// Current accumulated bar (for nT > 1).
1135    acc: Option<Candlestick>,
1136    /// Number of ticks accumulated into `acc`.
1137    tick_count: usize,
1138    /// Buffered items ready to be yielded on the next poll.
1139    pending: std::collections::VecDeque<Result<CandlestickItem, InternalProviderError>>,
1140}
1141
1142impl<S> SynthesizeTicksStream<S> {
1143    fn new(raw: S, n: usize) -> Self {
1144        Self {
1145            raw,
1146            n,
1147            history_ended: false,
1148            acc: None,
1149            tick_count: 0,
1150            pending: std::collections::VecDeque::new(),
1151        }
1152    }
1153}
1154
1155impl<S> futures_util::Stream for SynthesizeTicksStream<S>
1156where
1157    S: futures_util::Stream<Item = Result<TickItem, InternalProviderError>>,
1158{
1159    type Item = Result<CandlestickItem, InternalProviderError>;
1160
1161    fn poll_next(
1162        self: std::pin::Pin<&mut Self>,
1163        cx: &mut std::task::Context<'_>,
1164    ) -> std::task::Poll<Option<Self::Item>> {
1165        use std::task::Poll;
1166
1167        let mut this = self.project();
1168
1169        loop {
1170            if let Some(item) = this.pending.pop_front() {
1171                // Drain any buffered items first.
1172                return Poll::Ready(Some(item));
1173            }
1174
1175            // Poll the inner tick stream.
1176            let tick_item = match this.raw.as_mut().poll_next(cx) {
1177                Poll::Pending => return Poll::Pending,
1178                Poll::Ready(None) => {
1179                    // Inner stream ended. Flush partial accumulator.
1180                    if *this.n > 1
1181                        && let Some(bar) = this.acc.take()
1182                    {
1183                        return Poll::Ready(Some(Ok(CandlestickItem::Bar(bar))));
1184                    }
1185                    return Poll::Ready(None);
1186                }
1187                Poll::Ready(Some(item)) => item,
1188            };
1189
1190            match tick_item {
1191                Err(e) => return Poll::Ready(Some(Err(e))),
1192                Ok(TickItem::HistoryEnd) => {
1193                    // Flush any partial accumulated bar before the boundary.
1194                    if *this.n > 1
1195                        && let Some(bar) = this.acc.take()
1196                    {
1197                        this.pending.push_back(Ok(CandlestickItem::Bar(bar)));
1198                        *this.tick_count = 0;
1199                    }
1200                    *this.history_ended = true;
1201                    this.pending.push_back(Ok(CandlestickItem::HistoryEnd));
1202                }
1203                Ok(TickItem::Tick(tick)) => {
1204                    if *this.n == 1 {
1205                        // 1T: each tick is its own bar.
1206                        this.pending
1207                            .push_back(Ok(CandlestickItem::Bar(Candlestick::from_tick(tick))));
1208                    } else {
1209                        // nT: accumulate ticks. ask/bid are not meaningful for
1210                        // multi-tick bars (only defined for 1T), so they stay NaN.
1211                        match this.acc {
1212                            None => {
1213                                let mut bar = Candlestick::from_tick(tick);
1214                                bar.ask = f64::NAN;
1215                                bar.bid = f64::NAN;
1216                                *this.acc = Some(bar);
1217                                *this.tick_count = 1;
1218                            }
1219                            Some(bar) => {
1220                                bar.high = bar.high.max(tick.price);
1221                                bar.low = bar.low.min(tick.price);
1222                                bar.close = tick.price;
1223                                bar.volume += tick.volume;
1224                                *this.tick_count += 1;
1225                            }
1226                        }
1227
1228                        if *this.history_ended {
1229                            // Realtime: emit the current partial bar on every tick.
1230                            if let Some(bar) = *this.acc {
1231                                this.pending.push_back(Ok(CandlestickItem::Bar(bar)));
1232                            }
1233                            // Reset accumulator once n ticks are complete.
1234                            if *this.tick_count >= *this.n {
1235                                *this.acc = None;
1236                                *this.tick_count = 0;
1237                            }
1238                        } else {
1239                            // History: emit only after n ticks are accumulated.
1240                            if *this.tick_count >= *this.n {
1241                                if let Some(bar) = this.acc.take() {
1242                                    this.pending.push_back(Ok(CandlestickItem::Bar(bar)));
1243                                }
1244                                *this.tick_count = 0;
1245                            }
1246                        }
1247                    }
1248                }
1249            }
1250        }
1251    }
1252}
1253
1254/// Synthesizes a [`CandlestickItem`] stream from a raw tick stream.
1255fn synthesize_ticks<S>(raw: S, n: usize) -> SynthesizeTicksStream<S>
1256where
1257    S: futures_util::Stream<Item = Result<TickItem, InternalProviderError>>,
1258{
1259    SynthesizeTicksStream::new(raw, n)
1260}
navi_vm/data_provider.rs

navi_vm/
data_provider.rs
