基于RxJS实现值偏差压力检测：输出阈值外的变化百分比

我有一个Observable，它会生成快速的无限正整数序列。需要将其转换为值变化百分比序列，仅当在指定时间窗口（毫秒级）内，变化百分比超出设定阈值时才输出。

代码示例 #

function getPressurePercents(input: Observable<number>, minChange: number, iw: number) {
   // 返回一个Observable，仅当在iw时间窗口内值变化百分比超出minChange时，输出该变化百分比
}

const obs = getPressurePercents(input, 0.5, 3000).subscribe(val => {
    // 输入值在过去3秒内变化了至少0.5%
    // val = 变化百分比，Math.abs(val) >= 0.5
    // 正值表示百分比上升，负值表示百分比下降
});

需求细节 #

在上述示例中，目标是得到一个Observable：当输入数值在3秒窗口内出现0.5%的变化/偏差时，输出对应的变化百分比。

该需求用于识别值变化的压力峰值，例如股票价格的快速大幅波动（价格峰值），并在变化超出阈值时触发通知。

重要说明 #

实现需同时支持上升和下降两种场景：窗口内的百分比下降需输出负百分比。

我实现了一个可直接集成到RxJS管道的操作符，相比纯RxJS内置操作符组合的方案更简洁易懂，具体代码如下：

import { Observable, scan, filter, map } from 'rxjs';

function getPressurePercents(input: Observable<number>, minChange: number, windowMs: number): Observable<number> {
  return input.pipe(
    scan((acc, current) => {
      // 记录当前时间与值，同时清理窗口外的历史数据
      const now = Date.now();
      acc.history = acc.history.filter(entry => now - entry.timestamp <= windowMs);
      acc.history.push({ timestamp: now, value: current });

      // 取窗口内最早的有效值作为基准
      const initialValue = acc.history[0]?.value;
      if (!initialValue) return acc;

      // 计算变化百分比
      acc.changePercent = ((current - initialValue) / initialValue) * 100;
      return acc;
    }, { history: [] as { timestamp: number; value: number }[], changePercent: 0 }),
    // 过滤出符合阈值要求的变化
    filter(state => Math.abs(state.changePercent) >= minChange),
    // 仅输出变化百分比数值
    map(state => state.changePercent)
  );
}

方案说明 #

• 用scan维护时间窗口内的历史值集合，自动剔除超出窗口的旧数据
• 以窗口内第一个有效值为基准，计算当前值的变化百分比
• 通过filter筛选出满足阈值条件的结果，最终输出纯百分比数值
• 天然支持正负百分比输出：上升为正，下降为负

内容的提问来源于stack exchange，提问作者vitaly-t