c# - 使用基于时间戳的 ReactiveX 实现移动平均

Question

我正在使用 Rx.Net 并且我有 Observable 发出时间序列点( double ，时间戳)。每次新点到达时，我想计算过去 30 秒的平均值。我想我需要某种不是基于计数而是基于时间戳的重叠窗口/缓冲区。

我找到了 this主题与 SlidingWindow 实现，但我不知道如何适应我的问题。

编辑:

感谢 this我了解到我可以使用 Scan 运算符并缓冲我的点，所以基本上这解决了问题。但也许有更好的方法来做到这一点？

Answer 1

Buffer和 Window向前看，你想要一些可以回顾的东西。 Scan是最好的起点:

public static IObservable<List<T>> BackBuffer<T>(this IObservable<T> source, TimeSpan ts)
{
    return BackBuffer(source, ts, Scheduler.Default);
}
public static IObservable<List<T>> BackBuffer<T>(this IObservable<T> source, TimeSpan ts, IScheduler scheduler)
{
    return source
        .Timestamp()
        .Scan(new List<Timestamped<T>>(), (list, element) => list
            .Where(ti => scheduler.Now - ti.Timestamp <= ts)
            .Concat(Enumerable.Repeat(element, 1))
            .ToList()
        )
        .Select(list => list.Select(t => t.Value).ToList());
}

一旦您拥有 BackBuffer ，或类似的东西，那么剩下的就变得简单了:

source
    .BackBuffer(TimeSpan.FromMilliseconds(70))
    .Select(list => list.Average())
    .Subscribe(average => Console.WriteLine(average));