# 流式指标 (/zh/features/indicators/streaming-indicators)

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

```python title="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  # (1)
>>> streamed = []
>>> for i, price in enumerate(close.values):
...     if i == 0:
...         weighted_avg = price
...     state = vbt.nb.ewm_mean_acc_nb(  # (2)
...         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
```

1.  EMA 的完整状态只有三个数，从一次更新传递至下一次。
2.  一次更新：输入状态包含 K 线索引、新价格、先前状态和设置。

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

```python title="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)  # (1)
>>> 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
```

1.  用于测量单次更新耗时的代表性状态。两种函数均已预先调用一次，因此不计编译耗时。

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

## 流式指标适合哪些场景 \[#where-streaming-indicators-help]

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

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

## 流式状态 \[#streaming-state]

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

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

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

## 累加器构建块 \[#accumulator-building-blocks]

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

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

## Python、Numba 与 Rust \[#python-numba-and-rust]

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

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

!!! info "教程"
    会员专属的[超快 SuperTrend（英文）](https://members.vectorbt.pro/tutorials/superfast-supertrend/streaming/)教程使用累加器构建流式
    SuperTrend，并与批量版本比较。

## 流式指标 \[#streaming-indicators]

新增于 v2026.9.5

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

=== "Python / Numba"
    ```python title="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(  # (1)
    ...         vbt.ind_enums.OBVAIS(close, prev_close, volume, obv)
    ...     )
    ...     prev_close, obv = close, out.cumsum  # (2)
    ...     print(out.value)
    10.0
    30.0
    25.0
    ```

    1.  累加器也可以在 Numba 编译循环或回调内运行。
    2.  将前收盘价和累计能量潮传入下一次更新。

=== "Rust"
    ```rust title="Update Bollinger Bands and RSI for each price"
    use vectorbtpro_rust::error::VbtResult;
    use vectorbtpro_rust::indicators::streaming::{Bbands, Rsi};

    fn main() -> VbtResult<()> {
        let mut bands = Bbands::builder().window(3).build()?;
        let mut rsi = Rsi::builder().window(3).build()?;
        for price in [100.0, 101.0, 102.0, 99.0, 98.0, 103.0] {
            let band = bands.update(price); // (1)
            let strength = rsi.update(price).rsi;
            if strength.is_finite() { // (2)
                println!("${price:.0}: middle={:.2}, RSI={strength:.2}", band.middle);
            }
        }
        Ok(())
    }
    ```

    1.  结果还包含布林带上轨和下轨。
    2.  在 RSI 拥有足够观测满足三根 K 线窗口前，跳过输出。

    ```text title="Output"
    $99: middle=100.67, RSI=40.00
    $98: middle=99.67, RSI=30.77
    $103: middle=100.00, RSI=74.65
    ```

!!! info "教程"
    更多内容请参阅[实时模拟（英文）](https://members.vectorbt.pro/tutorials/from-python-to-rust/live/)教程。

## 累加器 \[#accumulators]

新增于 1.0.10

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

```python title="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
```