LINQ 计算 SortedList<dateTime,double> 的移动平均值答案

【问题标题】：LINQ to calculate a moving average of a SortedList<dateTime,double>LINQ 计算 SortedList<dateTime,double> 的移动平均值
【发布时间】：2011-07-07 04:57:39
【问题描述】：

我有一个SortedList<dateTime,double> 形式的时间序列。我想计算这个系列的移动平均值。我可以使用简单的 for 循环来做到这一点。我想知道是否有更好的方法来使用 linq。

我的版本：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace ConsoleApplication1
{
    class Program
    {
        static void Main(string[] args)
        {
            var mySeries = new SortedList<DateTime, double>();
            mySeries.Add(new DateTime(2011, 01, 1), 10);
            mySeries.Add(new DateTime(2011, 01, 2), 25);
            mySeries.Add(new DateTime(2011, 01, 3), 30);
            mySeries.Add(new DateTime(2011, 01, 4), 45);
            mySeries.Add(new DateTime(2011, 01, 5), 50);
            mySeries.Add(new DateTime(2011, 01, 6), 65);

            var calcs = new calculations();
            var avg = calcs.MovingAverage(mySeries, 3);
            foreach (var item in avg)
            {
                Console.WriteLine("{0} {1}", item.Key, item.Value);                
            }
        }
    }
    class calculations
    {
        public SortedList<DateTime, double> MovingAverage(SortedList<DateTime, double> series, int period)
        {
            var result = new SortedList<DateTime, double>();

            for (int i = 0; i < series.Count(); i++)
            {
                if (i >= period - 1)
                {
                    double total = 0;
                    for (int x = i; x > (i - period); x--)
                        total += series.Values[x];
                    double average = total / period;
                    result.Add(series.Keys[i], average);  
                }

            }
            return result;
        }
    }
}

【问题讨论】：

我会在迁移到 LINQ 之前对其进行测试。通常一个简单的手写 for 循环会在性能上击败 LINQ。
经过测试，手工编码的非Linq解决方案确实是一个更好（阅读速度更快）的解决方案

标签： c# linq moving-average

【解决方案1】：

为了实现 O(n) 的渐近性能（就像手动编码的解决方案一样），您可以使用 Aggregate 函数，如

series.Skip(period-1).Aggregate(
  new {
    Result = new SortedList<DateTime, double>(), 
    Working = List<double>(series.Take(period-1).Select(item => item.Value))
  }, 
  (list, item)=>{
     list.Working.Add(item.Value); 
     list.Result.Add(item.Key, list.Working.Average()); 
     list.Working.RemoveAt(0);
     return list;
  }
).Result;

累积值（实现为匿名类型）包含两个字段：Result 包含到目前为止建立的结果列表。 Working 包含最后一个 period-1 元素。聚合函数将当前值添加到工作列表中，构建当前平均值并将其添加到结果中，然后从工作列表中删除第一个（即最旧的）值。

“种子”（即累积的起始值）是通过将第一个 period-1 元素放入 Working 并将 Result 初始化为一个空列表来构建的。

因此，聚合从元素 period 开始（通过在开头跳过 (period-1) 元素）

在函数式编程中，这是聚合（或fold）函数的典型使用模式，顺便说一句。

两个备注：

解决方案不是“功能上”干净的，因为相同的列表对象（Working 和Result）在每个步骤中都被重用。如果某些未来的编译器尝试自动并行化 Aggregate 函数，我不确定这是否会导致问题（另一方面，我也不确定，这是否可能……）。纯功能解决方案应该在每一步都“创建”新列表。

另请注意，C# 缺乏强大的列表表达式。在一些假设的 Python-C# 混合伪代码中，可以编写聚合函数，如

(list, item)=>
  new {
    Result = list.Result + [(item.Key, (list.Working+[item.Value]).Average())], 
    Working=list.Working[1::]+[item.Value]
  }

在我看来会更优雅一点:)

【讨论】：

【解决方案2】：

对于使用 LINQ 计算移动平均线的最有效的方法，您不应该使用 LINQ！

相反，我建议创建一个帮助类，以最有效的方式（使用循环缓冲区和因果移动平均过滤器）计算移动平均，然后是扩展方法 使其可供 LINQ 访问。

首先是移动平均线

public class MovingAverage
{
    private readonly int _length;
    private int _circIndex = -1;
    private bool _filled;
    private double _current = double.NaN;
    private readonly double _oneOverLength;
    private readonly double[] _circularBuffer;
    private double _total;

    public MovingAverage(int length)
    {
        _length = length;
        _oneOverLength = 1.0 / length;
        _circularBuffer = new double[length];
    }       

    public MovingAverage Update(double value)
    {
        double lostValue = _circularBuffer[_circIndex];
        _circularBuffer[_circIndex] = value;

        // Maintain totals for Push function
        _total += value;
        _total -= lostValue;

        // If not yet filled, just return. Current value should be double.NaN
        if (!_filled)
        {
            _current = double.NaN;
            return this;
        }

        // Compute the average
        double average = 0.0;
        for (int i = 0; i < _circularBuffer.Length; i++)
        {
            average += _circularBuffer[i];
        }

        _current = average * _oneOverLength;

        return this;
    }

    public MovingAverage Push(double value)
    {
        // Apply the circular buffer
        if (++_circIndex == _length)
        {
            _circIndex = 0;
        }

        double lostValue = _circularBuffer[_circIndex];
        _circularBuffer[_circIndex] = value;

        // Compute the average
        _total += value;
        _total -= lostValue;

        // If not yet filled, just return. Current value should be double.NaN
        if (!_filled && _circIndex != _length - 1)
        {
            _current = double.NaN;
            return this;
        }
        else
        {
            // Set a flag to indicate this is the first time the buffer has been filled
            _filled = true;
        }

        _current = _total * _oneOverLength;

        return this;
    }

    public int Length { get { return _length; } }
    public double Current { get { return _current; } }
}

此类提供了一个非常快速和轻量级的 MovingAverage 过滤器实现。它创建一个长度为 N 的循环缓冲区，并为每个附加的数据点计算一个加法、一个减法和一个乘法，这与蛮力实现的每个点的 N 次乘加相反。

接下来，对它进行 LINQ 验证！

internal static class MovingAverageExtensions
{
    public static IEnumerable<double> MovingAverage<T>(this IEnumerable<T> inputStream, Func<T, double> selector, int period)
    {
        var ma = new MovingAverage(period);
        foreach (var item in inputStream)
        {
            ma.Push(selector(item));
            yield return ma.Current;
        }
    }

    public static IEnumerable<double> MovingAverage(this IEnumerable<double> inputStream, int period)
    {
        var ma = new MovingAverage(period);
        foreach (var item in inputStream)
        {
            ma.Push(item);
            yield return ma.Current;
        }
    }
}

上述扩展方法包装了 MovingAverage 类并允许插入到 IEnumerable 流中。

现在开始使用！

int period = 50;

// Simply filtering a list of doubles
IEnumerable<double> inputDoubles;
IEnumerable<double> outputDoubles = inputDoubles.MovingAverage(period);   

// Or, use a selector to filter T into a list of doubles
IEnumerable<Point> inputPoints; // assuming you have initialised this
IEnumerable<double> smoothedYValues = inputPoints.MovingAverage(pt => pt.Y, period);

【讨论】：

谢谢，强大的 for 循环嘲笑 .Zip.Scan.Select(Tuple) 方法！
几年后，但确实是一个可靠的方法。

【解决方案3】：

您已经有了一个答案，告诉您如何使用 LINQ，但坦率地说，我不会在这里使用 LINQ，因为与您当前的解决方案相比，它的性能很可能很差，而且您现有的代码已经很清楚了.

但是，您可以保留一个运行总数并在每次迭代时调整它，而不是计算每个步骤中先前 period 元素的总数。也就是说，改变这个：

total = 0;
for (int x = i; x > (i - period); x--)
    total += series.Values[x];

到这里：

if (i >= period) {
    total -= series.Values[i - period];
}
total += series.Values[i];

这意味着无论period 的大小如何，您的代码都将花费相同的时间来执行。

【讨论】：

这并不是真正回答问题。 OP 想知道如何在 Linq 中做到这一点。
在我看来，不使用 LINQ 是该问题的有效答案。 LINQ 很棒，但在这里用错了工具。
其实我只是想知道如何做好。那就是说。以后，我可能会直接从 SQL DB 中提取这些值。在这种情况下，全 LINQ 解决方案可能会更好。我会对它们进行基准测试，看看哪个更快。

【解决方案4】：

这个区块

double total = 0;
for (int x = i; x > (i - period); x--)
    total += series.Values[x];
double average = total / period;

可以改写为：

double average = series.Values.Skip(i - period + 1).Take(period).Sum() / period;

您的方法可能如下所示：

series.Skip(period - 1)
    .Select((item, index) =>
        new 
        {
            item.Key,            
            series.Values.Skip(index).Take(period).Sum() / period
        });

如您所见，linq 非常具有表现力。我建议从Introducing LINQ 和101 LINQ Samples 之类的教程开始。

【讨论】：

注意 O(n^2) 的运行时间，因为您需要在每一步跳过越来越多的元素（并且 afaik Skip(i) 必须调用 @987654327 @ i 次）。请在 O(n) 时间查看我对解决方案的回复...（我刚刚注意到下面的 OP 评论说他/她将来可能会从 SQL 数据库中获取值。在这种情况下我强烈反对这个解决方案！）
@Andre 欢迎您。 @MartinStettner 是的，你是对的。我尝试编写最优雅的解决方案，而不是最有效的......

【解决方案5】：

要以更实用的方式执行此操作，您需要一个存在于 Rx 但不存在于 LINQ 中的 Scan 方法。

让我们看看如果我们有一个扫描方法会是什么样子

var delta = 3;
var series = new [] {1.1, 2.5, 3.8, 4.8, 5.9, 6.1, 7.6};

var seed = series.Take(delta).Average();
var smas = series
    .Skip(delta)
    .Zip(series, Tuple.Create)
    .Scan(seed, (sma, values)=>sma - (values.Item2/delta) + (values.Item1/delta));
smas = Enumerable.Repeat(0.0, delta-1).Concat(new[]{seed}).Concat(smas);

这是扫描方法，取自here：

public static IEnumerable<TAccumulate> Scan<TSource, TAccumulate>(
    this IEnumerable<TSource> source,
    TAccumulate seed,
    Func<TAccumulate, TSource, TAccumulate> accumulator
)
{
    if (source == null) throw new ArgumentNullException("source");
    if (seed == null) throw new ArgumentNullException("seed");
    if (accumulator == null) throw new ArgumentNullException("accumulator");

    using (var i = source.GetEnumerator())
    {
        if (!i.MoveNext())
        {
            throw new InvalidOperationException("Sequence contains no elements");
        }
        var acc = accumulator(seed, i.Current);

        while (i.MoveNext())
        {
            yield return acc;
            acc = accumulator(acc, i.Current);
        }
        yield return acc;
    }
}

这应该比brute force method 具有更好的性能，因为我们使用运行总计来计算 SMA。

这是怎么回事？

首先，我们需要计算我们称之为seed 的第一个周期。然后，我们根据累积的种子值计算每个后续值。为此，我们需要旧值（即 t-delta）和我们将序列压缩在一起的最新值，一次从头开始，一次移动 delta。

最后，我们通过为第一个周期的长度添加零并添加初始种子值来进行一些清理。

【讨论】：

刚看到这个。很有意思！必须尝试一下，看看它是否在 C# for i 循环上有所改进
@AndreP。除了比蛮力更有效之外，这些值是以惰性方式计算的。所以假设你有 200k 个值，但是只写 smas.Take(1000)，它只会计算前 1000 个移动平均值。
在阅读了问题（而不是所有的答案）之后，我只是设计了相同的东西（尽管我调用了我的函数AggregateSeq）

【解决方案6】：

另一种选择是使用MoreLINQ的Windowed方法，大大简化了代码：

var averaged = mySeries.Windowed(period).Select(window => window.Average(keyValuePair => keyValuePair.Value));

【讨论】：

【解决方案7】：

我使用这段代码来计算 SMA：

private void calculateSimpleMA(decimal[] values, out decimal[] buffer)
{
    int period = values.Count();     // gets Period (assuming Period=Values-Array-Size)
    buffer = new decimal[period];    // initializes buffer array
    var sma = SMA(period);           // gets SMA function
    for (int i = 0; i < period; i++)
        buffer[i] = sma(values[i]);  // fills buffer with SMA calculation
}

static Func<decimal, decimal> SMA(int p)
{
    Queue<decimal> s = new Queue<decimal>(p);
    return (x) =>
    {
        if (s.Count >= p)
        {
            s.Dequeue();
        }
        s.Enqueue(x);
        return s.Average();
    };
}

【讨论】：

【解决方案8】：

这是一个扩展方法：

public static IEnumerable<double> MovingAverage(this IEnumerable<double> source, int period)
{
    if (source is null)
    {
        throw new ArgumentNullException(nameof(source));
    }

    if (period < 1)
    {
        throw new ArgumentOutOfRangeException(nameof(period));
    }

    return Core();

    IEnumerable<double> Core()
    {
        var sum = 0.0;
        var buffer = new double[period];
        var n = 0;
        foreach (var x in source)
        {
            n++;
            sum += x;
            var index = n % period;
            if (n >= period)
            {
                sum -= buffer[index];
                yield return sum / period;
            }

            buffer[index] = x;
        }
    }
}

【讨论】：