RXJS 的价值偏差压力

Question

我有一个

Observable

，它可以生成正整数值的快速无限序列。

并且我需要以某种方式将其替换为值变化百分比序列，无论何时超出变化百分比阈值，同时在一定的间隔窗口内（以毫秒为单位）。

代码示例：

function getPressurePercents(input: Observable, minChange: number, iw: number) {
   // return an observable that emits value-change percents that are outside
   // minChange percent value during any iw (interval window)
}

const obs = getPressurePercents(input, 0.5, 3000).subscribe(val => {
    // values in the input have changed by at least 0.5% in the last 3 seconds
    // value = value-change percent, Math.abs(value) >= 0.5
    // it is positive for value-percent increases, and negative for decreases
});

在上面的示例中，我希望获得一个可观察量，只要输入数字在任何 3 秒窗口内显示 0.5% 的变化/偏差，它就会发出值变化百分比。

这样做的目的是发现价值变化压力峰值，并收到超出阈值的所有百分比的通知。

重要提示：它必须同时向上和向下工作，即窗口内的百分比下降预计会相应地发出负百分比。

Answer 1

我相信这会做你想做的事。请注意，我使用 rxjs 中的 bufferTime、map、filter 和pairwise。今年我试图回馈一些时间来帮助别人。希望这对您有帮助：

import { Observable } from 'rxjs';
import { bufferTime, map, filter, pairwise } from 'rxjs/operators';

function getPressurePercents(input, minChange, iw) {
    return new Observable(subscriber => {
        input.pipe(
            bufferTime(iw), // Buffer values for each interval window
            map(arr => {
                if (arr.length < 2) return null; // Ensure at least two values to compare
                const start = arr[0];
                const end = arr[arr.length - 1];
                const percentChange = ((end - start) / start) * 100;
                return percentChange;
            }),
            filter(change => change !== null && Math.abs(change) >= minChange) // Filter by minChange
        ).subscribe({
            next: (value) => {
                subscriber.next(value);
            },
            error: (err) => subscriber.error(err),
            complete: () => subscriber.complete()
        });
    });
}


https://rxjs.dev/api/index/function/bufferTime
https://rxjs.dev/api/index/function/filter 
https://rxjs.dev/api/operators/pairwise