数据提供者

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` 流协议

可用方法

示例

自定义内存提供者示例

异步数据库提供者示例

参阅