資料提供者

DataProvider 向 VM 提供 K 線資料和標的元資料，服務於兩個目的：

主 chart 資料 — Instance::run() 從中拉取主 K 線流驅動逐 bar 執行。
request.security() 資料 — 多週期和跨標的呼叫也從同一 provider 拉取。

未設定 provider 時，run() 立即返回不執行任何 bar，所有 request.security() 呼叫均回傳 na。

將其作為 Instance::builder() 的第一個參數傳入，然後呼叫 run()：

rust

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

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

Trait 定義

rust

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

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

    fn candlesticks(
        &self,
        symbol: String,
        timeframe: TimeFrame,
        from_time: i64,
        count: Option<usize>,
    ) -> Result<
        impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
        DataProviderError<Self::Error>,
    >;

    // 可選——實作此方法以支援 tick 時間框架（1T、2T、…）
    fn ticks(
        &self,
        symbol: String,
        from_time: i64,
        count: Option<usize>,
    ) -> Result<
        impl Stream<Item = Result<TickItem, DataProviderError<Self::Error>>> + 'static,
        DataProviderError<Self::Error>,
    > {
        Ok(stream::empty()) // 預設：無 tick 資料
    }

    // --- 輔助數據流（全部可選，預設傳回空流）---

    fn currency_rate(&self, from: Currency, to: Currency, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;

    fn financial(&self, symbol: String, financial_id: String, period: String,
        currency: Option<Currency>, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;

    fn earnings(&self, symbol: String, field: EarningsField,
        currency: Option<Currency>, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;

    fn dividends(&self, symbol: String, field: DividendsField,
        currency: Option<Currency>, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;

    fn economic(&self, country_code: String, field: String, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;

    fn splits(&self, symbol: String, field: SplitsField, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;

    fn data(&self, function: String, args: DataArgs, from_time: i64)
        -> Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>;
}

symbol_info 使用 async_trait(?Send)——直接實作為 async fn 即可。所有傳回串流的方法（candlesticks、ticks、currency_rate 等）傳回 Result<impl Stream<Item = ...> + 'static, DataProviderError<Self::Error>>——標的無效等情況應直接傳回 Err（參見下方錯誤類型）。成功時傳回 Ok(stream)，stream 不能借用 self，在傳回前將所有需要的共享狀態 clone 進去。簡單場景可以直接傳回迭代器流；有分支邏輯時，內部使用 Box::pin() 即可。

所有傳回串流的方法均支援有限串流和無限串流兩種模式：

有限串流 — 串流產出所有資料後終止。適用於歷史回測場景。若串流結束時仍有未確認的即時 bar/資料點，VM 會自動確認它。
無限串流 — 串流永不終止，持續在 HistoryEnd 之後產出新資料。適用於實盤交易和即時監控場景。

兩種模式使用相同的 HistoryEnd 邊界協議：HistoryEnd 之前的資料以阻塞方式載入；HistoryEnd 之後以非阻塞方式輪詢，避免某個停滯的即時資料源凍結整體執行。所有輔助方法預設傳回空流。

`candlesticks`

返回指定 (symbol, timeframe) 的 CandlestickItem 串流。count 在 request 串流需要最近視窗時為 Some(n)，主圖執行時為 None。

兩種模式使用相同的 CandlestickItem 協議。

`CandlestickItem`

CandlestickItem 是一個列舉，包含兩個變體：

CandlestickItem::Bar(candle) — 一根 K 線，其角色（歷史 bar 或即時 bar）由其在串流中相對於 HistoryEnd 的位置決定。HistoryEnd 之前的所有 bar 必須是已完全收盤的歷史 bar。HistoryEnd 之後的 Bar 為即時 bar——時間戳可能與最後一根歷史 bar 相同（若查詢時該週期仍在形成中），也可能是更晚的新時間戳（新 bar）。時間戳相同的連續 item 表示原地更新（tick 更新）；時間戳更晚的 item 會隱式確認上一個 bar 並開啟新 bar。
CandlestickItem::HistoryEnd — 恰好發送一次，在所有已確認的歷史 bar 發送完畢後、即時更新開始之前發送。DataProvider 必須確保 HistoryEnd 之前的每一根 bar 都是已完全收盤的週期。若當前週期仍在形成中，請在 HistoryEnd 之後將其作為即時 Bar item 推送。若串流結束時存在未確認的即時 bar，VM 會自動確認它。

串流協議時序

期望的 item 順序如下：

Bar(t0)  Bar(t1)  …  Bar(tN)   ← 所有已確認的歷史 bar（全部已收盤）
HistoryEnd                      ← 邊界標記（發送一次）
Bar(tA)                         ← 第一根即時 bar
                                   （若查詢時該週期仍在形成中，tA 可能等於 tN）
Bar(tA)                         ← 相同時間戳 → 原地更新
Bar(tB)  (tB > tA)              ← 更晚的時間戳：
                                   VM 自動確認 tA，開啟 tB
Bar(tB)                         ← 更新 tB
…                               ← 有限串流：串流在此結束
                                   無限串流：持續產出

當即時 bar 收盤時，不要單獨發送一個 Bar 來顯式確認它。只需發送時間戳更晚的下一個 Bar(tB)，VM 會在內部自動處理確認邏輯。

要求	說明
升序時間	K 線必須按 `time` 嚴格升序返回
起點早於 `from_time`	從第一根圖表 K 線開始時刻（epoch 毫秒）之前的一根 K 線開始，讓 VM 能在第一個 chart bar 之前初始化 MTF 狀態
尊重 `count`	當 `count` 為 `Some(n)` 時，僅返回該請求串流最近的 `n` 根歷史 bar
`HistoryEnd`	在所有已確認的歷史 bar 發送完畢後、即時更新開始之前 emit；`HistoryEnd` 之前的所有 bar 必須是已完全收盤的
`'static` stream	返回的 stream 不能借用 `self`——在返回前 clone 需要的句柄
`InvalidSymbol`	不支援該標的時，從方法本身傳回 `Err(DataProviderError::InvalidSymbol(_))`（而非作為串流中的第一個 item）
其他錯誤	網路故障等使用 `Err(DataProviderError::Other(e))`——可從方法傳回，也可作為串流中的 item

錯誤類型

當唯一可能的失敗是 InvalidSymbol 時，使用 type Error = std::convert::Infallible：

rust

type Error = std::convert::Infallible;

`ignore_invalid_symbol`

當 candlesticks 返回 Err(DataProviderError::InvalidSymbol(_)) 時：

如果 Navi 腳本以 ignore_invalid_symbol = true 呼叫了 request.security()，該呼叫每個 bar 都靜默返回 na。
否則 VM 拋出運行時錯誤。

這樣可以區分「標的不在資料集中」（軟錯誤，可忽略）和「連線失敗」（硬錯誤，必須回報）：

rust

fn candlesticks(
    &self,
    symbol: String,
    timeframe: TimeFrame,
    _from_time: i64,
    _count: Option<usize>,
) -> Result<
    impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
    DataProviderError<Self::Error>,
> {
    if !self.supports(&symbol, timeframe) {
        // 軟錯誤：腳本可透過 ignore_invalid_symbol = true 選擇接收 na 而非報錯
        return Err(DataProviderError::InvalidSymbol(
            format!("{symbol}/{timeframe} not found"),
        ));
    }
    // ... 正常路徑
    Ok(my_stream)
}

`ticks`

傳回指定標的的 TickItem 串流。實作此方法以支援 tick 時間框架（1T、2T、…）。當圖表或 request.security() 呼叫使用 tick 時間框架時，VM 會呼叫 ticks() 而非 candlesticks()，並在內部自動合成 K 線。

預設實作傳回空串流（不執行任何 bar）。與 candlesticks 相同，此串流支援有限和無限兩種模式。

`TickItem`

TickItem 是一個列舉，包含兩個變體：

TickItem::Tick(tick) — 單一市場 tick。Tick 結構體包含 time（epoch 毫秒）、price、volume、ask 和 bid。
TickItem::HistoryEnd — 恰好發送一次，位於所有歷史 tick 之後、即時 tick 之前。

K 線合成規則

VM 根據所請求的時間框架數量 n 從 tick 串流合成 CandlestickItem::Bar：

時間框架	規則	`ask` / `bid`
`1T`	每個 tick → 一根 bar；`open = high = low = close = price`	保留自 tick
`nT`（n > 1）	每 `n` 個 tick → 一根確認 bar；OHLCV 累積	始終為 `na`

對於 nT，若 HistoryEnd 到達時累加器中存在未滿的部分 bar（tick 數 < n），該部分 bar 會在 HistoryEnd 傳播之前先行 flush。

ask 和 bid 內建變數僅在 1T 圖表上有意義，在其他所有時間框架上均傳回 na。

範例

rust

fn ticks(
    &self,
    symbol: String,
    from_time: i64,
    count: Option<usize>,
) -> Result<
    impl Stream<Item = Result<TickItem, DataProviderError<Self::Error>>> + 'static,
    DataProviderError<Self::Error>,
> {
    let items = self.load_ticks(&symbol, from_time, count);
    Ok(async_stream::stream! {
        for tick in items {
            yield Ok(TickItem::Tick(tick));
        }
        yield Ok(TickItem::HistoryEnd);
        // 繼續 yield TickItem::Tick(...) 以提供即時 tick
    })
}

`symbol_info`

返回指定標的的元資料。所有欄位都是 Option——只填寫已知的部分，VM 會用交易所前綴的預設值補全其餘欄位。

沒有額外元資料時，直接返回 PartialSymbolInfo::default()：

rust

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

利用 `fields` 優化資料查詢

fields 是編譯期靜態分析得到的 bitset，標識腳本實際用到了哪些 syminfo.* property。你可以用它跳過腳本從未讀取的欄位的昂貴查詢；也可以直接忽略它、始終返回全部欄位——VM 只會使用它需要的部分。

rust

async fn symbol_info(
    &self,
    symbol: String,
    fields: SyminfoFields,
) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
    let mut info = PartialSymbolInfo::default();

    // 基礎價格元資料——代價低，始終獲取
    let meta = self.pool.query_basic(&symbol).await
        .map_err(DataProviderError::Other)?;
    info.min_move    = Some(meta.min_move);
    info.price_scale = Some(meta.price_scale);
    info.currency    = Some(meta.currency);

    // 分析師資料——遠端呼叫代價高，按需跳過
    if fields.intersects(
        SyminfoFields::RECOMMENDATIONS_BUY
        | SyminfoFields::RECOMMENDATIONS_SELL
        | SyminfoFields::RECOMMENDATIONS_HOLD
        | SyminfoFields::TARGET_PRICE_AVERAGE,
    ) {
        let rec = self.pool.query_analyst_data(&symbol).await
            .map_err(DataProviderError::Other)?;
        info.recommendations_buy    = rec.buy;
        info.recommendations_sell   = rec.sell;
        info.target_price_average   = rec.target_avg;
    }

    Ok(info)
}

完整欄位列表

欄位	SyminfoFields flag	說明
`market`	—	標的所屬交易所/市場（`Option<Market>`）。變體：`NYSE`、`NASDAQ`、`SHSE`、`SZSE`、`HKEX`、`SGX`
`description`	`DESCRIPTION`	標的可讀名稱，如 `"Apple Inc."`
`type_`	`TYPE`	工具類型（`Option<SymbolType>`），見下方變體列表
`country`	`COUNTRY`	交易所所在國家的 ISO 3166-1 alpha-2 代碼，如 `"US"`、`"HK"`
`isin`	`ISIN`	國際證券識別號（ISIN）
`root`	`ROOT`	衍生品根代碼，如 `"ESZ4"` 的根代碼為 `"ES"`
`min_move`	`MIN_MOVE`, `MIN_TICK`	mintick 公式分子：`mintick = min_move / price_scale`
`price_scale`	`PRICE_SCALE`, `MIN_TICK`	mintick 公式分母。如 `min_move=1, price_scale=100` → mintick `0.01`
`point_value`	`POINT_VALUE`	合約點值乘數，通常為 `1.0`；如 ES 期貨為 `50.0`
`currency`	`CURRENCY`	標的價格所用貨幣（`Option<Currency>`），見下方變體列表
`expiration_date`	`EXPIRATION_DATE`	當前期貨合約到期日
`current_contract`	`CURRENT_CONTRACT`	標的合約的代碼識別符
`base_currency`	`BASE_CURRENCY`	外匯/加密貨幣對的基礎貨幣，如 `BTCUSD` 中的 `BTC`
`employees`	`EMPLOYEES`	員工人數（僅限股票）
`industry`	`INDUSTRY`	行業分類
`sector`	`SECTOR`	板塊分類
`min_contract`	`MIN_CONTRACT`	最小合約規模
`volume_type`	`VOLUME_TYPE`	成交量解釋方式：`Base`（基礎貨幣）、`Quote`（計價貨幣）、`Tick`（成交筆數）
`shareholders`	`SHAREHOLDERS`	股東人數
`shares_outstanding_float`	`SHARES_OUTSTANDING_FLOAT`	流通股數量
`shares_outstanding_total`	`SHARES_OUTSTANDING_TOTAL`	總股本數量
`recommendations_buy`	`RECOMMENDATIONS_BUY`	分析師買入評級數量
`recommendations_buy_strong`	`RECOMMENDATIONS_BUY_STRONG`	分析師強烈買入評級數量
`recommendations_hold`	`RECOMMENDATIONS_HOLD`	分析師持有評級數量
`recommendations_sell`	`RECOMMENDATIONS_SELL`	分析師賣出評級數量
`recommendations_sell_strong`	`RECOMMENDATIONS_SELL_STRONG`	分析師強烈賣出評級數量
`recommendations_date`	`RECOMMENDATIONS_DATE`	最新分析師評級更新日期
`recommendations_total`	`RECOMMENDATIONS_TOTAL`	分析師評級總數量
`target_price_average`	`TARGET_PRICE_AVERAGE`	分析師平均目標價
`target_price_date`	`TARGET_PRICE_DATE`	最新目標價更新日期
`target_price_estimates`	`TARGET_PRICE_ESTIMATES`	目標價預測的總數量
`target_price_high`	`TARGET_PRICE_HIGH`	分析師最高目標價
`target_price_low`	`TARGET_PRICE_LOW`	分析師最低目標價
`target_price_median`	`TARGET_PRICE_MEDIAN`	分析師中位目標價

完整變體列表請參閱 Rust API 文件中的 SymbolType 和 Currency 列舉定義。

標的格式

標的使用 "EXCHANGE:TICKER" 格式。VM 將 Navi 腳本中的原始字串直接傳遞給 provider，不做任何修改：

Navi 腳本	傳給 provider 的值
`syminfo.tickerid`	`"NASDAQ:AAPL"`（來自圖表標的）
`"BINANCE:BTCUSDT"`	`"BINANCE:BTCUSDT"`

Ticker 表達式（如 "NASDAQ:AAPL*0.5+NYSE:SPY*0.5"）會被 VM 分解為各個標的的獨立查詢，再分別傳給 provider。

TimeFrame

TimeFrame 直接來自 Navi 腳本的週期字串參數：

Navi 字串	含義
`"1"`	1 分鐘
`"5"`	5 分鐘
`"60"`	60 分鐘
`"D"`	日線
`"W"`	週線
`"M"`	月線
`"1T"`	1 tick（每個 tick 一根 bar；`ask`/`bid` 可用）
`"2T"`	2 tick（每 2 個 tick 合成一根 bar）

Vec<Candlestick>

Vec<Candlestick> 直接實作了 DataProvider。對於單標的離線回測，將 vec 直接傳給 Instance::builder()，無需任何包裝：

rust

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

let bars = vec![
    Candlestick::new(
        1_700_000_000_000, // epoch 毫秒
        150.0, 155.0, 149.0, 153.0,
        1_000_000.0, 0.0,
        TradeSession::Regular,
    ),
    // ... 更多按時間升序排列的 K 線
];

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

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

對於惰性載入或串流資料，先收集為 Vec<Candlestick>：

rust

// 從迭代器收集後再建構
let bars: Vec<Candlestick> = load_bars_from_disk().collect();
let mut instance = Instance::builder(bars, source, TimeFrame::days(1), "NASDAQ:AAPL")
    .build().await?;

如果只需要腳本元資料（輸入、腳本類型等），可使用獨立函數 script_info()——無需資料提供者、品種或週期：

rust

use navi_vm::script_info;

let info = script_info(source)?;

輔助數據流

除了 K 線和品種資訊外，DataProvider 還提供可選方法用於提供輔助數據，供 request.currency_rate()、request.financial()、request.earnings()、 request.dividends()、request.economic()、request.splits() 和 request.data() 使用。

所有方法都有預設的空流實作，因此您只需覆寫數據源支援的方法。

`AuxDataItem` 流協議

輔助流產出 AuxDataItem 值，遵循與 K 線流相同的協議（同樣支援有限串流和無限串流）：

text

Data(t0)  Data(t1)  …  Data(tN)   ← 歷史數據點
HistoryEnd                         ← 分界標記
Data(tA)  Data(tB)  …             ← 可選的即時更新

在收到 HistoryEnd 之前，VM 會急切地排空流（阻塞）以建構完整的歷史數據集。收到 HistoryEnd 之後，條目以非阻塞方式消費。

可用方法

currency_rate(from, to, from_time) — 以串流方式提供帶時間戳的匯率數據點，使 VM 能在逐 bar 執行時解析 request.currency_rate()。strategy.convert_to_account() / strategy.convert_to_symbol() 進行跨貨幣回測時也會在內部消費此串流。每個 AuxDataPoint.value 為一單位 from 兌換 to 的匯率。

financial(symbol, financial_id, period, currency, from_time) — 以串流方式提供週期性財務指標值，VM 將其映射到圖表 bar 以供 request.financial() 使用。financial_id 為 provider 自訂的指標鍵（如 "TOTAL_REVENUE"），period 為報告週期（"FQ" / "FY"）， currency 請求幣種轉換。數據點通常稀疏——每個披露日一條。

earnings(symbol, field, currency, from_time) — 以串流方式提供逐事件的盈利數據，供 request.earnings() 使用。field 為 EarningsField 列舉，選擇 provider 應傳回的數值類型（Actual、Estimate 或 Standardized）。VM 在原始串流基礎上施加 gaps 和 lookahead 邏輯。

dividends(symbol, field, currency, from_time) — 以串流方式提供逐事件的股息數據，供 request.dividends() 使用。field 為 DividendsField 列舉，選擇 Gross（稅前）或 Net（稅後）。與 earnings 一樣，由 VM 進行 gaps / lookahead 處理。

economic(country_code, field, from_time) — 以串流方式提供帶時間戳的宏觀經濟指標值，供 request.economic() 使用。country_code 為 ISO 3166-1 alpha-2 國家代碼，field 為 provider 自訂的指標鍵。

splits(symbol, field, from_time) — 以串流方式提供拆股比例分量，供 request.splits() 使用。field 為 SplitsField 列舉，選擇 Numerator（分子）或 Denominator（分母）。與 earnings / dividends 一樣，由 VM 進行 gaps / lookahead 處理。

data(function, args, from_time) — 路由來自 Navi 腳本 request.data() 呼叫的自訂數據請求。 function 是 provider 自訂的路由鍵，標識要獲取的數據集。args 是由 Navi 腳本傳入的 map<string, any> 參數建構的 DataArgs 映射；使用帶 #[derive(serde::Deserialize)] 的結構體配合 args.deserialize::<T>() 實現類型安全的參數提取。透過與其他輔助方法相同的 AuxDataItem 協議，將 series float 數據點對齊到圖表 bar 上傳回。

每個方法傳回 Result<impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static, DataProviderError<Self::Error>>。當標識符無法識別時，從方法本身傳回 Err(DataProviderError::InvalidSymbol(_))，VM 會按與 K 線流相同的方式處理（ignore_invalid_symbol / ignore_invalid_currency）。

範例

rust

fn currency_rate(
    &self,
    from: Currency,
    to: Currency,
    _from_time: i64,
) -> Result<
    impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
    DataProviderError<Self::Error>,
> {
    let pool = Arc::clone(&self.pool);
    Ok(async_stream::stream! {
        match pool.query_fx_rates(from, to).await {
            Err(e) => yield Err(DataProviderError::Other(e)),
            Ok(rates) => {
                for rate in rates {
                    yield Ok(AuxDataItem::Data(AuxDataPoint {
                        time: rate.timestamp_ms,
                        value: rate.rate,
                    }));
                }
                yield Ok(AuxDataItem::HistoryEnd);
            }
        }
    })
}

rust

fn data(
    &self,
    function: String,
    args: DataArgs,
    from_time: i64,
) -> Result<
    impl Stream<Item = Result<AuxDataItem, DataProviderError<Self::Error>>> + 'static,
    DataProviderError<Self::Error>,
> {
    #[derive(serde::Deserialize)]
    struct MyMetricParams {
        symbol: String,
        period: i64,
        adjusted: bool,
    }

    match function.as_str() {
        "my_metric" => {
            let params = args
                .deserialize::<MyMetricParams>()
                .map_err(|e| DataProviderError::Other(MyError::from(e)))?;
            Ok(Box::pin(self.fetch_my_metric(params, from_time)))
        }
        _ => Ok(Box::pin(futures_util::stream::empty())),
    }
}

對應的 Navi 腳本：

navi

// 無參數
vix = request.data("vix")

// 傳入類型化參數
args = map.new<string, any>()
args.put("symbol", "AAPL")
args.put("period", 14)
args.put("adjusted", true)
val = request.data("my_metric", args)

自訂內存提供者示例

需要多標的支援（如 request.security()）時，直接實作 DataProvider：

rust

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

use futures_util::{Stream, stream};
use navi_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,
        _fields: SyminfoFields,
    ) -> Result<PartialSymbolInfo, DataProviderError<Self::Error>> {
        Ok(PartialSymbolInfo::default())
    }

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

        // 從 from_time 前一根 K 線開始，確保 VM 能初始化 MTF 狀態
        let start = bars
            .partition_point(|b| b.time < from_time)
            .saturating_sub(1);

        let sliced = bars[start..].to_vec();
        let recent = if let Some(count) = count {
            let start = sliced.len().saturating_sub(count);
            sliced[start..].to_vec()
        } else {
            sliced
        };

        Ok(stream::iter(
            recent
                .into_iter()
                .map(|c| Ok(CandlestickItem::Bar(c)))
                .chain(std::iter::once(Ok(CandlestickItem::HistoryEnd))),
        ))
    }
}

使用方式：

rust

let mut provider = InMemoryProvider { data: HashMap::new() };
provider.data.insert(
    ("NASDAQ:AAPL".to_string(), TimeFrame::weeks(1)),
    vec![/* ... */],
);

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

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

異步資料庫提供者示例

適用於從遠端 API 或資料庫獲取資料的場景：

rust

use std::sync::Arc;
use futures_util::Stream;

struct DatabaseProvider {
    pool: Arc<DbPool>,
}

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

    async fn symbol_info(
        &self,
        symbol: String,
        _fields: SyminfoFields,
    ) -> 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,
        count: Option<usize>,
    ) -> Result<
        impl Stream<Item = Result<CandlestickItem, DataProviderError<Self::Error>>> + 'static,
        DataProviderError<Self::Error>,
    > {
        let pool = Arc::clone(&self.pool);
        Ok(async_stream::stream! {
            match pool.query_candles(&symbol, timeframe, from_time, count).await {
                Err(e) => yield Err(DataProviderError::Other(e)),
                Ok(rows) => {
                    for row in rows {
                        yield Ok(CandlestickItem::Bar(row.into_candlestick()));
                    }
                    // 發出歷史結束標記，讓 VM 切換為非阻塞輪詢
                    yield Ok(CandlestickItem::HistoryEnd);
                }
            }
        })
    }
}

參閱

執行腳本 — 使用資料提供者執行 VM
安全與執行限制 — max_security_depth 和 max_security_calls 限制

脚本管理

標注

附錄

資料提供者

Trait 定義

`candlesticks`

`CandlestickItem`

串流協議時序

錯誤類型

`ignore_invalid_symbol`

`ticks`

`TickItem`

K 線合成規則

範例

`symbol_info`

利用 `fields` 優化資料查詢

完整欄位列表

標的格式

TimeFrame

Vec<Candlestick>

輔助數據流

`AuxDataItem` 流協議

可用方法

範例

自訂內存提供者示例

異步資料庫提供者示例

參閱

資料提供者 ​

Trait 定義 ​

candlesticks ​

CandlestickItem ​

串流協議時序 ​

錯誤類型 ​

ignore_invalid_symbol ​

ticks ​

TickItem ​

K 線合成規則 ​

範例 ​

symbol_info ​

利用 fields 優化資料查詢 ​

完整欄位列表 ​

標的格式 ​

TimeFrame ​

Vec<Candlestick> ​

輔助數據流 ​

AuxDataItem 流協議 ​

可用方法 ​

範例 ​

自訂內存提供者示例 ​

異步資料庫提供者示例 ​

參閱 ​

資料提供者

Trait 定義

`candlesticks`

`CandlestickItem`

串流協議時序

錯誤類型

`ignore_invalid_symbol`

`ticks`

`TickItem`

K 線合成規則

範例

`symbol_info`

利用 `fields` 優化資料查詢

完整欄位列表

標的格式

TimeFrame

Vec<Candlestick>

輔助數據流

`AuxDataItem` 流協議

可用方法

範例

自訂內存提供者示例

異步資料庫提供者示例

參閱