c# - RX : Stateful transform of sequence, 例如指数移动平均线

Question

如何在 RX 中对序列进行简单的状态转换？

假设我们要对 IObservable noisySequence 进行指数移动平均变换。

每当 noisySequence tick 时，emaSequence 应该 tick 并返回值(previousEmaSequenceValue*(1-lambda) + latestNoisySequenceValue*lambda)

我猜我们使用了 Subjects，但究竟如何呢？

    public static void Main()
    {

        var rand = new Random();

        IObservable<double> sequence  = Observable
            .Interval(TimeSpan.FromMilliseconds(1000))
            .Select(value => value + rand.NextDouble());

        Func<double, double> addNoise = x => x + 10*(rand.NextDouble() - 0.5);

        IObservable<double> noisySequence = sequence.Select(addNoise);

        Subject<double> exponentialMovingAverage = new Subject<double>(); // ??? 


        sequence.Subscribe(value => Console.WriteLine("original sequence "+value));
        noisySequence.Subscribe(value => Console.WriteLine("noisy sequence " + value));
        exponentialMovingAverage.Subscribe(value => Console.WriteLine("ema sequence " + value));

        Console.ReadLine();
    }

Answer 1

这是将状态附加到序列的方法。在本例中，它计算最后 10 个值的平均值。

var movingAvg = noisySequence.Scan(new List<double>(),
(buffer, value)=>
{
    buffer.Add(value);
    if(buffer.Count>MaxSize)
    {
        buffer.RemoveAt(0);
    }
    return buffer;
}).Select(buffer=>buffer.Average());

但是您也可以使用 Window(Buffer 是其概括)来获取平均值。

noisySequence.Window(10)
   .Select(window=>window.Average())
   .SelectMany(averageSequence=>averageSequence);