Python 与 Rust 流式指标

指标能否在每根新 K 线到来时更新，而无需重算全部历史？VBT 使用累加器在 Python 中计算流式指标：小型函数接受先前状态与一个新值，并返回下一状态。它们与批量版本使用相同公式，因此使用相同输入、设置和初始化时，结果可与批量计算一致，无需每次更新都重算完整历史。

Update an EMA one bar at a time and compare it with the batch result

close = vbt.YFData.pull("BTC-USD", start="2024-01-01", end="2025-01-01").close
span = 20
alpha = 2 / (span + 1)

old_wt, weighted_avg, nobs = 1.0, np.nan, 0  
streamed = []
for i, price in enumerate(close.values):
    if i == 0:
        weighted_avg = price
    state = vbt.nb.ewm_mean_acc_nb(  
        vbt.nb.EWMMeanAIS(i, price, old_wt, weighted_avg, nobs, alpha, 1, True)
    )
    old_wt, weighted_avg, nobs = state.old_wt, state.weighted_avg, state.nobs
    streamed.append(state.value)

batch = close.vbt.ewm_mean(span, minp=1)
print(np.allclose(streamed, batch), round(streamed[-1], 2), round(batch.iloc[-1], 2))

True 96421.8 96421.8

流式值在每根 K 线上都与批量 EMA 匹配。保留状态而非历史，才能让更新成本保持恒定：

Time one streaming update against recomputing the history

import timeit

in_state = vbt.nb.EWMMeanAIS(1, 100.0, 1.0, 100.0, 1, alpha, 1, True)  
rows = {}
for n in [1_000, 100_000]:
    arr = np.random.default_rng(42).normal(size=n).cumsum() + 1000
    batch_us = min(timeit.repeat(
        lambda: vbt.nb.ewm_mean_1d_nb(arr, span, 1, True), number=100, repeat=5
    )) / 100 * 1e6
    stream_us = min(timeit.repeat(
        lambda: vbt.nb.ewm_mean_acc_nb(in_state), number=10000, repeat=5
    )) / 10000 * 1e6
    rows[n] = {
        "recompute history (µs)": round(batch_us, 1),
        "one streaming update (µs)": round(stream_us, 1),
    }
pd.DataFrame(rows).T

        recompute history (µs)  one streaming update (µs)
1000                       4.9                        0.8
100000                   450.8                        0.9

在 Apple M3 上测量。重新计算的成本随历史增长，而流式更新保持在一微秒以下，其中大部分还是 Python 调用开销。

流式指标适合哪些场景

研究任务	累加器如何帮助
每根已完成 K 线到来时更新信号	从保留状态计算下一个值
跟踪多个标的或参数设置	为每次计算保留独立状态，而非重跑各自完整历史
在模拟回调中计算指标	随策略逐 K 线决策一起更新指标
以连续批次处理历史	在批次间传递状态，让计算从相同起点继续

由您的数据流程或模拟提供观测。例如，将每根已完成 K 线输入 EMA 和 RSI，组合输出为入场条件，再传给策略。数据流水线介绍数据采集和更新，实时模拟介绍如何随观测到来推进投资组合。

流式状态

每个流式指标都由状态定义。EMA 的状态是运行均值及其权重，滚动均值是运行总和及即将离开窗口的值，RSI 则是平均涨幅和跌幅。保存状态后，指标可以稍后、在另一进程或新数据到来后继续，无需重放此前计算的完整历史。请将设置和当前位置与状态一同保留。对于 Numba 滚动计算，还应保留观测离开窗口时需要的近期值。Rust 累加器在内部管理这些缓冲区。

每个标的和参数设置都有自己的状态。例如，您可以跟踪多个资产的快慢均线，而不会混合各自历史。

流式指标仍需要预热。滚动窗口在观测不足时返回 NaN，指数均值也取决于起点：一年前开始的 EMA 与昨天开始的 EMA，今天会给出不同值。两者必须一致时，请以相同设置从与批量计算相同的位置开始流式计算。先输入较早 K 线，也可在希望交易或评估的区间前完成预热，详见前向滚动与交叉验证。

累加器构建块

累加器涵盖滚动和、均值、标准差、z 分数、最小和最大值、指数均值与标准差、交叉，以及 RSI、ADX、能量潮和 SuperTrend 等指标。由于每个累加器都仅使用当前输入和保留状态或窗口值，因此多个累加器可以在一次数据遍历中组合，从简单指标构建复杂指标。下方累加器功能介绍将其中两个组合为一次遍历的滚动 z 分数。

流式处理让每次更新专注于新观测，因此累加器特别适合长历史、多标的和模拟循环内部计算。您可以将其与其余指标的批量计算组合。上方 EMA 基准测试在同一数据上展示两种方法。

Python、Numba 与 Rust

同样的累加器可在 Python 循环、Numba 编译循环或模拟回调，以及 Rust 中运行。Rust 版本管理自己的滚动缓冲区，因此 Rust 程序只需输入新值，如下方流式指标功能介绍所示。这让您可以在 Python 中制作原型，再将相同逻辑转移到编译的实时系统，无需改变数学计算。

Rust 累加器还可克隆，从同一预热状态分支。向每份副本输入不同未来观测，即可比较情景，无需重放共同历史。使用 reset() 以相同配置开始全新计算。它们是自定义指标、原生研究程序和实时计算的实用构建块。

教程

会员专属的超快 SuperTrend（英文）教程使用累加器构建流式 SuperTrend，并与批量版本比较。

流式指标

新增于v2026.9.5

✅ 让指标随市场持续更新。VBT 的 Numba 和 Rust 累加器每次更新一个观测，使用与批量执行相同的公式。Rust 累加器甚至管理自己的滚动缓冲区，让您输入新数据后直接得到下一个值。

Update on-balance volume for each bar

prev_close = np.nan
obv = 0.0
for close, volume in [(100.0, 10.0), (102.0, 20.0), (101.0, 5.0)]:
    out = vbt.ind_nb.obv_acc_nb(  
        vbt.ind_enums.OBVAIS(close, prev_close, volume, obv)
    )
    prev_close, obv = close, out.cumsum  
    print(out.value)

10.0
30.0
25.0

教程

更多内容请参阅实时模拟（英文）教程。

累加器

新增于1.0.10

✅ 大多数用 Pandas 和 NumPy 实现的滚动指标需要多次遍历数据。例如，计算三个数组之和至少需要两次遍历。如果想逐 K 线迭代计算这类指标，就必须预先计算全部结果并存入内存，或重新计算每个窗口，这可能显著影响性能。而累加器维护内部状态，让新数据点到来时即可计算指标值，从而实现尽可能高的性能。

Design a one-pass rolling z-score

@njit
def fastest_rolling_zscore_1d_nb(arr, window, minp=None, ddof=1):
    if minp is None:
        minp = window
    out = np.full(arr.shape, np.nan)
    cumsum = 0.0
    cumsum_sq = 0.0
    nancnt = 0

    for i in range(len(arr)):
        pre_window_value = arr[i - window] if i - window >= 0 else np.nan
        mean_in_state = vbt.nb.RollMeanAIS(
            i, arr[i], pre_window_value, cumsum, nancnt, window, minp
        )
        mean_out_state = vbt.nb.rolling_mean_acc_nb(mean_in_state)
        _, _, _, mean = mean_out_state
        std_in_state = vbt.nb.RollStdAIS(
            i, arr[i], pre_window_value, cumsum, cumsum_sq, nancnt, window, minp, ddof
        )
        std_out_state = vbt.nb.rolling_std_acc_nb(std_in_state)
        cumsum, cumsum_sq, nancnt, _, std = std_out_state
        out[i] = (arr[i] - mean) / std
    return out

data = vbt.YFData.pull("BTC-USD")
rolling_zscore = fastest_rolling_zscore_1d_nb(data.returns.values, 14)
data.symbol_wrapper.wrap(rolling_zscore)

Date
2014-09-17 00:00:00+00:00         NaN
2014-09-18 00:00:00+00:00         NaN
2014-09-19 00:00:00+00:00         NaN
                                  ...
2023-02-01 00:00:00+00:00    0.582381
2023-02-02 00:00:00+00:00   -0.705441
2023-02-03 00:00:00+00:00   -0.217880
Freq: D, Name: BTC-USD, Length: 3062, dtype: float64

(data.returns - data.returns.rolling(14).mean()) / data.returns.rolling(14).std()

Date
2014-09-17 00:00:00+00:00         NaN
2014-09-18 00:00:00+00:00         NaN
2014-09-19 00:00:00+00:00         NaN
                                  ...
2023-02-01 00:00:00+00:00    0.582381
2023-02-02 00:00:00+00:00   -0.705441
2023-02-03 00:00:00+00:00   -0.217880
Freq: D, Name: Close, Length: 3062, dtype: float64

流式指标适合哪些场景

流式状态

累加器构建块

Python、Numba 与 Rust

流式指标

累加器

本页目录