带有Rx.NET的信号滤波器

Question

我想在Rx.NET中实现一个信号滤波器，它从一组初始系数开始。随着时间的推移，必须根据观测数据值的快照重新计算滤波器系数。

这里有一个小原型，展示了它应该如何工作。为了简单起见，我选择滤波器长度和用于重新计算滤波器系数的历史数据值的数量相同（示例中为3）。

该示例使用bufferedAt10中的副作用来重新计算系数。这不是我想要的。

在实际应用中，数据以不规则的时间步长出现，系数应在特定时间每天更新一次或每周更新一次。我可以很容易地延长缓冲区，但我如何让系统运行，并以一种干净的函数方式改变观测器的滤波器系数？

       // create a hot obvervable to produce data
        const int bufLen = 3;
        var rng = new Random();
        var period = TimeSpan.FromSeconds(0.5);
        var observable = Observable.Interval(period)
        .Select(i => new {Time = DateTime.Now, Price = rng.NextDouble()})
        .Do(e => Console.WriteLine("original : {0}", e))
        .Publish();
        observable.Connect();
        Console.WriteLine("Press any key to subscribe");
        Console.ReadKey();
        // buffer of length bufLen used for filter calculation (every tick) and filter 
        // coefficient update (at a lower frequency)
        var buffered = observable.Buffer(bufLen, 1);
        // apply the signal filter with coefficients in `coeff`
        var coeff = new List<Double>() {1.0, 1.0, 1.0};  // these will be updated on the way from new data 
        var filtered = buffered.Select(e =>
        {
            var f = 0.0;
            for (var i = 0; i < bufLen; i++)
            {
                f += e[i].Price*coeff[i]; // apply the filter with coefficients `coeff`
            }
            return new {Time = DateTime.Now, FilteredPrice = f};
        });
        var s1 = filtered.Subscribe(e => Console.WriteLine("filtered : {0} (coeff {1},{2},{3})", e, coeff[0], coeff[1], coeff[2]));
        // recalculate the filter coefficients say every 10 seconds 
        var bufferedAt10 = buffered.DistinctUntilChanged(e => (e[bufLen - 1].Time.TimeOfDay.Seconds / 10) * 10);
        var s2 = bufferedAt10.Subscribe(e =>
        {
            Console.WriteLine("recalc of coeff : {0}", e[bufLen - 1].Time);
            for (var i = 0; i < bufLen; i++)
            {
                // a prototypical function that takes the buffer and uses it to "recalibrate" the filter coefficients
                coeff[i] = coeff[i] + e[bufLen - 1 - i].Price;
            }
            Console.WriteLine("updated coeffs to {0},{1},{2}", coeff[0], coeff[1], coeff[2]);
        });

谢谢你的好建议。

Answer 1

以下内容未经测试，但我认为它应该满足您的需求。其背后的想法是，分流流对一个流进行系数更新，然后将它们与WithLatestFrom重新组合在一起。我使用Sample和Scan来执行周期"调整"。您的自定义时间戳可以通过使用TimeStamp运算符来完成。您也可以考虑将Publish向下移动到Buffer之后，否则您将有两个流生成缓冲区，但这取决于您。

const int bufLen = 3;
var rng = new Random();
var period = TimeSpan.FromSeconds(0.5);
var observable = Observable.Interval(period)
  .Select( => rng.NextDouble())
  .Publish();
observable.Connect();
Console.WriteLine("Press any key to subscribe");
Console.ReadKey();
var buffered = observable.Buffer(bufLen, 1);
var seed = new [] {1.0, 1.0, 1.0};
var coefficients = buffered
                     //Samples for a new value every 10 seconds
                     .Sample(TimeSpan.FromSeconds(10))
                     //Updates the seed value and emits it after every update
                     .Scan(seed, 
                       //Use good old fashion Linq
                       (coeff, delta) => coeff.Zip(delta.Reverse(), 
                                         (c, d) => c + d.Price)
                                         .ToArray()
                       );
//Emits a new value everytime buffer emits, and combines it with the latest
//values from the coefficients Observable
//Kick off coefficients with the seed otherwise you need to wait 10 seconds 
//for the first value.
buffer.WithLatestFrom(coefficients.StartWith(seed), (e, coeff) => {
  return e.Zip(coeff, (x, c) => x.Price * c).Sum();
})
.TimeStamp()
.Subscribe(e => Console.WriteLine("filtered : {0}", e);