Data Providers

A DataProvider supplies candlestick history and symbol metadata to the VM. It serves two purposes:

1. Main chart data — Instance::run() pulls the primary K-line stream from it to drive bar-by-bar execution.
2. request.security() data — MTF and cross-symbol calls also pull from the same provider.

Without a provider, Instance::run() returns immediately with no bars executed, and all request.security() calls return na.

Pass it as the first argument to Instance::builder(), then run:

let instance = Instance::builder(MyProvider::new(), source, timeframe, "NASDAQ:AAPL")
    .build().await?;

instance.run_to_end("NASDAQ:AAPL", timeframe).await?;

Trait Definition

#[async_trait(?Send)]
pub trait DataProvider {
    type Error: fmt::Display + fmt::Debug;

    async fn symbol_info(
        &self,
        symbol: String,
    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>>;

    fn candlesticks(
        &self,
        symbol: String,
        timeframe: TimeFrame,
        from_time: i64,
    ) -> LocalBoxStream<'static, Result<CandlestickItem, DataProviderError<Self::Error>>>;
}

symbol_info uses async_trait(?Send) — implement it as a plain async fn. candlesticks returns LocalBoxStream<'static, Result<CandlestickItem, ...>> — the stream must not borrow self; clone any shared state before returning.

candlesticks

Returns a stream of CandlestickItem values for the requested (symbol, timeframe) pair.

CandlestickItem is an enum with three variants:

• CandlestickItem::Confirmed(candle) — a fully closed bar. Only used for historical bars. Do not emit Confirmed to signal that a realtime bar has closed; instead, send the next Realtime item with a later timestamp — the VM confirms the previous bar automatically.
• CandlestickItem::Realtime(candle) — a bar that is still forming. Consecutive items with the same timestamp update the bar in place (tick update). An item with a later timestamp implicitly confirms the previous bar and opens a new one. At most one realtime bar is open at any time.
• CandlestickItem::HistoryEnd — emitted exactly once, after all historical Confirmed bars and before the first Realtime bar. If the stream ends with an open Realtime bar, the VM confirms it automatically.

Stream sequence

The expected item sequence is:

Confirmed(t0)  Confirmed(t1)  …  Confirmed(tN)   ← historical closed bars
HistoryEnd                                        ← boundary marker (emit once)
Realtime(tA)                                      ← forming bar, first tick
Realtime(tA)                                      ← same timestamp → update in place
Realtime(tB)   (tB > tA)                          ← later timestamp:
                                                     VM auto-confirms tA, opens tB
Realtime(tB)                                      ← update tB
…

Do not emit Confirmed(tA) when the realtime bar at tA closes. The VM handles confirmation internally when it receives the next Realtime(tB).

Requirement	Detail
Ascending order	Bars must be yielded in strictly ascending time order
Start before from_time	Begin at or one bar before from_time (epoch ms of the first chart bar) so the VM can seed MTF state before the first chart bar opens
HistoryEnd	Emit after all historical bars and before any Realtime bars
'static stream	The LocalBoxStream must not borrow self — clone handles before returning
InvalidSymbol	Yield Err(DataProviderError::InvalidSymbol(_)) as the first item when the symbol is not supported
Other errors	Use Err(DataProviderError::Other(e)) for network failures and similar

Error type

Use type Error = std::convert::Infallible when the only possible failure is InvalidSymbol:

type Error = std::convert::Infallible;

ignore_invalid_symbol

When candlesticks yields Err(DataProviderError::InvalidSymbol(_)):

• If the Pine script called request.security(..., ignore_invalid_symbol = true), the call silently returns na for every bar.
• Otherwise the VM raises a runtime error.

This lets you distinguish "symbol not in dataset" (soft, ignorable) from "connection failed" (hard, always an error):

fn candlesticks(
    &self,
    symbol: String,
    timeframe: TimeFrame,
    _from_time: i64,
) -> LocalBoxStream<'static, Result<CandlestickItem, DataProviderError<Self::Error>>> {
    if !self.supports(&symbol, timeframe) {
        // Soft: script can opt in to receiving na instead of an error
        return Box::pin(stream::iter(vec![Err(DataProviderError::InvalidSymbol(
            format!("{symbol}/{timeframe} not found"),
        ))]));
    }
    // ... normal path
}

symbol_info

Returns metadata for the given symbol string. All fields are Option — supply only what you know; the VM merges the rest with defaults derived from the exchange prefix.

Return PartialSymbolInfo::default() if you have no additional metadata:

async fn symbol_info(
    &self,
    _symbol: String,
) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
    Ok(PartialSymbolInfo::default())
}

All fields

Field	Description
market	Exchange/market (Option<Market>). Variants: NYSE, NASDAQ, SHSE, SZSE, HKEX, SGX
description	Human-readable name, e.g. "Apple Inc."
type_	Instrument type (Option<SymbolType>). See variants below
country	ISO 3166-1 alpha-2 country code of the exchange, e.g. "US", "HK"
isin	International Securities Identification Number
root	Root ticker for derivatives, e.g. "ES" for "ESZ4"
min_move	Numerator of mintick: mintick = min_move / price_scale
price_scale	Denominator of mintick. E.g. min_move=1, price_scale=100 → mintick 0.01
point_value	Point value multiplier. Usually 1.0; e.g. 50.0 for ES futures
currency	Price currency (Option<Currency>). See variants below
expiration_date	Expiration date of the current futures contract
current_contract	Ticker identifier of the underlying contract
base_currency	Base currency for forex/crypto pairs, e.g. BTC in BTCUSD
employees	Number of employees (stocks only)
industry	Industry classification
sector	Sector classification
min_contract	Minimum contract size
volume_type	Volume interpretation: Base (base currency), Quote (quote currency), Tick (trade count)
shareholders	Number of shareholders
shares_outstanding_float	Free-float shares outstanding
shares_outstanding_total	Total shares outstanding
recommendations_buy	Analyst buy recommendation count
recommendations_buy_strong	Analyst strong-buy recommendation count
recommendations_hold	Analyst hold recommendation count
recommendations_sell	Analyst sell recommendation count
recommendations_sell_strong	Analyst strong-sell recommendation count
recommendations_date	Date of latest analyst recommendations update
recommendations_total	Total number of analyst recommendations
target_price_average	Average analyst price target
target_price_date	Date of latest price target update
target_price_estimates	Number of price target estimates
target_price_high	Highest analyst price target
target_price_low	Lowest analyst price target
target_price_median	Median analyst price target

See the SymbolType and Currency enum definitions in the Rust API docs for all variants.

Symbol Format

Symbols use the "EXCHANGE:TICKER" format. The VM passes the exact string from the Pine script without modification:

Pine script	Passed to provider
syminfo.tickerid	"NASDAQ:AAPL" (from the chart symbol)
"BINANCE:BTCUSDT"	"BINANCE:BTCUSDT"

Ticker expressions (e.g. "NASDAQ:AAPL*0.5+NYSE:SPY*0.5") are decomposed into individual symbol lookups before being passed to the provider.

Timeframe Values

TimeFrame is passed directly from the Pine script's timeframe string argument:

Pine string	Meaning
"1"	1-minute
"5"	5-minute
"60"	60-minute
"D"	Daily
"W"	Weekly
"M"	Monthly

Vec<Candlestick>

Vec<Candlestick> implements DataProvider directly. For single-symbol offline backtests, pass the vector straight to Instance::builder() — no wrapper needed:

use openpine_vm::{Candlestick, TimeFrame, TradeSession};

let bars = vec![
    Candlestick::new(
        1_700_000_000_000, // epoch ms
        150.0, 155.0, 149.0, 153.0,
        1_000_000.0, 0.0,
        TradeSession::Regular,
    ),
    // ... more bars in ascending time order
];

let mut instance = Instance::builder(bars, source, TimeFrame::days(1), "NASDAQ:AAPL")
    .build()
    .await?;

instance.run_to_end("NASDAQ:AAPL", TimeFrame::days(1)).await?;

For lazy or streaming sources, collect into a Vec<Candlestick> first:

// Collect from an iterator before building
let bars: Vec<Candlestick> = load_bars_from_disk().collect();
let mut instance = Instance::builder(bars, source, TimeFrame::days(1), "NASDAQ:AAPL")
    .build().await?;

When you only need script metadata (inputs, script type, etc.), use the standalone script_info() function instead — it requires no data provider, symbol, or timeframe:

use openpine_vm::script_info;

let info = script_info(source)?;

Custom In-Memory Provider Example

When you need multi-symbol support (e.g. for request.security()), implement DataProvider directly:

use std::{collections::HashMap, convert::Infallible};

use futures_util::{stream, stream::LocalBoxStream};
use openpine_vm::{
    Candlestick, CandlestickItem, DataProvider, DataProviderError, PartialSymbolInfo, TimeFrame,
    TradeSession,
};

struct InMemoryProvider {
    data: HashMap<(String, TimeFrame), Vec<Candlestick>>,
}

#[async_trait::async_trait(?Send)]
impl DataProvider for InMemoryProvider {
    type Error = Infallible;

    async fn symbol_info(
        &self,
        _symbol: String,
    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
        Ok(PartialSymbolInfo::default())
    }

    fn candlesticks(
        &self,
        symbol: String,
        timeframe: TimeFrame,
        from_time: i64,
    ) -> LocalBoxStream<'static, Result<CandlestickItem, DataProviderError<Self::Error>>> {
        let bars = self
            .data
            .get(&(symbol.clone(), timeframe))
            .cloned()
            .unwrap_or_default();

        // Start one bar before from_time so the VM can seed MTF state.
        let start = bars
            .partition_point(|b| b.time < from_time)
            .saturating_sub(1);

        Box::pin(stream::iter(
            bars[start..]
                .to_vec()
                .into_iter()
                .map(|c| Ok(CandlestickItem::Confirmed(c)))
                .chain(std::iter::once(Ok(CandlestickItem::HistoryEnd))),
        ))
    }
}

Usage:

let mut provider = InMemoryProvider { data: HashMap::new() };
provider.data.insert(
    ("NASDAQ:AAPL".to_string(), TimeFrame::weeks(1)),
    vec![
        Candlestick::new(
            1_700_000_000_000, // epoch ms
            150.0, 155.0, 149.0, 153.0,
            1_000_000.0, 0.0,
            TradeSession::Regular,
        ),
        // ... more bars in ascending time order
    ],
);

let mut instance = Instance::builder(provider, source, TimeFrame::days(1), "NASDAQ:AAPL")
    .build()
    .await?;

instance.run_to_end("NASDAQ:AAPL", TimeFrame::days(1)).await?;

Async Database Provider Example

For providers that fetch data from a remote API or database:

use std::sync::Arc;
use futures_util::stream::LocalBoxStream;

struct DatabaseProvider {
    pool: Arc<DbPool>,
}

#[async_trait::async_trait(?Send)]
impl DataProvider for DatabaseProvider {
    type Error = DbError;

    async fn symbol_info(
        &self,
        symbol: String,
    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
        self.pool
            .query_symbol_info(&symbol)
            .await
            .map_err(DataProviderError::Other)
    }

    fn candlesticks(
        &self,
        symbol: String,
        timeframe: TimeFrame,
        from_time: i64,
    ) -> LocalBoxStream<'static, Result<CandlestickItem, DataProviderError<Self::Error>>> {
        let pool = Arc::clone(&self.pool);
        Box::pin(async_stream::stream! {
            match pool.query_candles(&symbol, timeframe, from_time).await {
                Err(e) => yield Err(DataProviderError::Other(e)),
                Ok(rows) => {
                    for row in rows {
                        // Confirmed for historical bars;
                        // yield Realtime for a forming bar if available
                        yield Ok(CandlestickItem::Confirmed(row.into_candlestick()));
                    }
                    // Signal end of history so the VM switches to non-blocking poll
                    yield Ok(CandlestickItem::HistoryEnd);
                }
            }
        })
    }
}

See Also

• Security & Limits — max_security_depth and max_security_calls limits
• Instance Builder — full builder API reference