如何在 RX 中对序列进行简单、有状态的转换?
假设我们想要对 IObservable 噪声序列进行指数移动平均变换。
每当noisySequence勾选时,emaSequence应该勾选并返回值
(前一个EmaSequenceValue*(1-lambda) + 最新NoisySequenceValue*lambda)
我想我们使用主题,但具体如何呢?
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();
}
这就是将状态附加到序列的方法。在本例中,它计算最后 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);
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)