如何以一般方式减少对子集的平均计算？

Question

编辑：由于似乎没有人阅读此链接的原始问题，所以让我在这里介绍一下它的概要。

正如其他人所问的那样，最初的问题是，在给定大量值的情况下，总和将超过 Double 的数据类型所持有的值，如何计算这些值的平均值。

有几个答案说要按组计算，比如取 50 和 50 个数字，然后计算这些组内的平均值，然后最后取所有这些组的平均值，然后将它们组合起来得到最终的平均值。

我的立场是，除非你能保证所有这些值都可以分成多个大小相同的集合，否则你不能使用这种方法。有人敢在这里问这个问题，为了提供答案，所以就在这里。

基本上，给定任意数量的值，其中：

• 我事先知道值的数量（但同样，如果你不知道，你的答案会有什么变化？`）
• 我无法收集所有数字，也无法对它们求和（对于您的编程语言中的普通数据类型而言，总和太大了）

如何计算平均值？

这里的问题的其余部分概述了如何拆分成相同大小的集合的方法以及存在的问题，但我真的很想知道如何做到这一点。

请注意，我对数学非常了解，知道在数学理论方面，计算 A[1..N]/N 的总和会给我一个平均值，让我们假设有原因它不那么简单，我需要拆分工作负载，并且值的数量不一定可以被 3、7、50、1000 或其他任何值整除。

换句话说，我所追求的解决方案必须是通用的。

---

从这个问题：

• What is a good solution for calculating an average where the sum of all values exceeds a double’s limits?

我的立场是，将工作负载分成几组是不好的，除非你能确保这些组的大小相等。

---

编辑：最初的问题是关于特定数据类型可以容纳的上限，因为他总结了很多数字（作为示例给出的计数是 10^9） ，数据类型无法容纳总和。由于这是原始解决方案中的一个问题，我假设（这是我的问题的先决条件，很抱歉错过了这一点）数字太大而无法给出任何有意义的答案。

所以，直接除以值的总数就行了。正常 SUM/COUNT 解决方案失败的最初原因是 SUM 会溢出，但我们假设，对于这个问题，SET-SET/SET-SIZE 会下溢，或者其他什么。

重要的是我不能简单地求和，我不能简单地除以总值。如果我不能做到这一点，我的方法是否有效，我可以做些什么来解决它？

---

让我概述一下问题。

假设您要计算数字 1 到 6 的平均值，但您不能（无论出于何种原因）通过对数字求和、对数字进行计数、然后将总和除以计数来实现。换句话说，你不能简单地做 (1+2+3+4+5+6)/6。

换句话说，SUM(1..6)/COUNT(1..6) 已经出局了。我们在这里不考虑 NULL（如数据库中的 NULL）。

该问题的几个答案暗示能够将被平均的数字分成若干组，例如 3、50 或 1000 个数字，然后为此计算一些数字，然后最后组合这些值以获得最终平均值。

我的立场是，这在一般情况下是不可能的，因为这会使一些数字，出现在最后一组中的数字，或多或少比前一组中的所有数字更有价值，除非你可以拆分所有数字将数字分成大小相等的集合。

例如，要计算 1-6 的平均值，您可以将其分成 3 个数字的集合，如下所示：

/ 1   2   3 \   / 4   5   6 \
| - + - + - | + | - + - + - |
\ 3   3   3 /   \ 3   3   3 /  <-- 3 because 3 numbers in the set
 ----------      -----------
      2               2        <-- 2 because 2 equally sized groups

这给了你这个：

      2               5
      -       +       - = 3.5
      2               2

（注意：（1+2+3+4+5+6）/6 = 3.5，所以这里是正确的）

但是，我的观点是，一旦无法将多个值拆分为多个相同大小的集合，这种方法就会崩溃。例如，包含素数值的序列 1-7 呢？

一种类似的方法，不会一次性对所有值求和，并计算所有值，是否可行？

那么，有这样的方法吗？如何计算满足以下条件的任意数量值的平均值：

1. 无论出于何种原因，我都无法采用正常的总和/计数方法
2. 我事先知道值的数量（如果我不知道，那会改变答案吗？）

Answer 1

好吧，假设您将三个数字相加并除以三，然后将两个数字相加并除以二。你能从中得到平均值吗？

x = (a + b + c) / 3
y = (d + e) / 2
z = (f + g) / 2

你想要

r = (a + b + c + d + e + f + g) / 7

等于

r = (3 * (a + b + c) / 3 + 2 * (d + e) / 2 + 2 * (f + g) / 2) / 7
r = (3 * x + 2 * y + 2 * z) / 7

当然，上面的两行都溢出了，但是由于除法是分配的，所以我们这样做

r = (3.0 / 7.0) * x + (2.0 / 7.0) * y + (2.0 / 7.0) * z

保证不会溢出，因为我将 x、y 和 z 乘以小于一的分数。

这是这里的基本点。我既不会事先将所有数字除以总数，也不会超过溢出。

所以...如果您继续添加到累加器，跟踪您添加了多少数字，并始终测试下一个数字是否会导致溢出，然后您可以获得部分平均值，并计算最终平均值.

不，如果你事先不知道这些值，它不会改变任何东西（前提是你可以在对它们求和时计算它们）。

这是一个 Scala 函数。它不是惯用的 Scala，因此更容易理解：

def avg(input: List[Double]): Double = {
  var partialAverages: List[(Double, Int)] = Nil
  var inputLength = 0
  var currentSum = 0.0
  var currentCount = 0
  var numbers = input

  while (numbers.nonEmpty) {
    val number = numbers.head
    val rest = numbers.tail
    if (number > 0 && currentSum > 0 && Double.MaxValue - currentSum < number) {
      partialAverages = (currentSum / currentCount, currentCount) :: partialAverages
      currentSum = 0
      currentCount = 0
    } else if (number < 0 && currentSum < 0 && Double.MinValue - currentSum > number) {
      partialAverages = (currentSum / currentCount, currentCount) :: partialAverages
      currentSum = 0
      currentCount = 0
    }
    currentSum += number
    currentCount += 1
    inputLength += 1
    numbers = rest
  }
  partialAverages = (currentSum / currentCount, currentCount) :: partialAverages

  var result = 0.0
  while (partialAverages.nonEmpty) {
    val ((partialSum, partialCount) :: rest) = partialAverages
    result += partialSum * (partialCount.toDouble / inputLength)
    partialAverages = rest
  }

  result
}

编辑： 不会乘以 2 和 3，让我回到“数据类型不支持”的范围？

没有。如果你最后潜水 7 点，绝对是。但是在这里，您是在总和的每一步进行除法。即使在您的实际情况下，重量（2/7 和 3/7）也将在可管理的数字范围内（例如 1/10 ~ 1/10000），与您的体重相比不会有很大的不同（即 @987654330 @)。

PS：我想知道为什么我要研究这个答案，而不是写我可以赚取代表的答案:-)

Answer 2

如果您事先知道值的数量（比如N），您只需添加1/N + 2/N + 3/N 等，假设您有值1, 2, 3。您可以将其拆分为任意数量的计算，然后将结果相加。这可能会导致精度略有下降，但这应该不是问题，除非您还需要超精确的结果。

如果您不提前知道项目的数量，您可能需要更有创意。但是您可以再次逐步进行。假设列表是1, 2, 3, 4。从mean = 1 开始。然后mean = mean*(1/2) + 2*(1/2)。然后mean = mean*(2/3) + 3*(1/3)。然后mean = mean*(3/4) + 4*(1/4)等。很容易概括，只需要确保括号中的数量是预先计算好的，以防止溢出。

当然，如果您想要极高的准确度（例如，超过 0.001% 的准确度），您可能需要比这更小心一点，否则应该没问题。

Answer 3

让X 成为您的样本集。以您喜欢的任何方式将其分成两组A 和B。定义delta = m_B - m_A，其中m_S 表示集合S 的平均值。那么

m_X = m_A + delta * |B| / |X|

其中|S| 表示集合S 的基数。现在您可以重复地将其应用于分区并计算平均值。

为什么这是真的？让s = 1 / |A| 和t = 1 / |B| 和u = 1 / |X|（为了符号方便）让aSigma 和bSigma 分别表示A 和B 中元素的总和，这样：

  m_A + delta * |B| / |X|
= s * aSigma + u * |B| * (t * bSigma - s * aSigma)
= s * aSigma + u * (bSigma - |B| * s * aSigma)
= s * aSigma + u * bSigma - u * |B| * s * aSigma
= s * aSigma * (1 - u * |B|) + u * bSigma
= s * aSigma * (u * |X| - u * |B|) + u * bSigma
= s * u * aSigma * (|X| - |B|) + u * bSigma
= s * u * aSigma * |A| + u * bSigma
= u * aSigma + u * bSigma
= u * (aSigma + bSigma)
= u * (xSigma)
= xSigma / |X|
= m_X

证明完成了。

从这里很明显，如何使用它来递归计算平均值（例如通过重复将集合分成两半）或如何使用它来并行计算集合的平均值。

著名的计算平均值的在线算法只是这种情况的一个特例。这是如果m 是{x_1, x_2, ... , x_n} 的平均值那么{x_1, x_2, ..., x_n, x_(n+1)} 的平均值是m + ((x_(n+1) - m)) / (n + 1) 的算法。所以用X = {x_1, x_2, ..., x_(n+1)}、A = {x_(n+1)}和B = {x_1, x_2, ..., x_n}我们恢复了在线算法。

Answer 4

跳出框框思考：改用中位数。计算起来要容易得多-那里有大量算法（例如使用队列），您通常可以构建很好的论据，说明为什么它对数据集更有意义（受极值影响较小等），并且您将遇到零问题数值精度。这将是快速和高效的。另外，对于大型数据集（听起来像您拥有），除非分布真的很奇怪，否则平均值和中位数的值将是相似的。

Answer 5

当您将数字分成几组时，您只是在除以总数还是我遗漏了什么？

你把它写成

/ 1   2   3 \   / 4   5   6 \
| - + - + - | + | - + - + - |
\ 3   3   3 /   \ 3   3   3 /
 ----------      -----------
      2               2

但这只是

/ 1   2   3 \   / 4   5   6 \
| - + - + - | + | - + - + - |
\ 6   6   6 /   \ 6   6   6 /

所以对于从 1 到 7 的数字，一个可能的分组只是

/ 1   2   3 \   / 4   5   6 \   / 7 \
| - + - + - | + | - + - + - | + | - |
\ 7   7   7 /   \ 7   7   7 /   \ 7 /

Answer 6

 

Average of x_1 .. x_N
    = (Sum(i=1,N,x_i)) / N
    = (Sum(i=1,M,x_i) + Sum(i=M+1,N,x_i)) / N
    = (Sum(i=1,M,x_i)) / N + (Sum(i=M+1,N,x_i)) / N

这可以重复应用，并且无论总和是否相等都是如此。所以：

• 继续添加术语，直到两个：
• • 添加另一个会溢出（或以其他方式丢失精度）
• • 除以 N 不会下溢
• 将总和除以 N
• 将结果添加到迄今为止的平均值中

有一个明显的尴尬情况，即序列末尾有一些非常小的项，这样在满足“除以 N 不会下溢”的条件之前，值就会用完。在这种情况下，只需丢弃这些值 - 如果它们对平均值的贡献不能用您的浮点类型表示，那么它尤其小于平均值的精度。因此，是否包含这些术语对结果没有任何影响。

还有一些不太明显的尴尬情况与个别求和的精度损失有关。例如，这些值的平均值是多少：

10^100, 1, -10^100

数学说它是 1，但浮点算术说它取决于你将各项相加的顺序，并且在 6 种可能性中有 4 种是 0，因为 (10^100) + 1 = 10^100。但我认为浮点算术的不可交换性是一个与这个问题不同且更普遍的问题。如果对输入进行排序是不可能的，我认为您可以做一些事情，即您可以维护许多不同大小的累加器，并将每个新值添加到其中任何一个将提供最佳精度的值。但我真的不知道。

Answer 7

这是另一种方法。您从某个来源一个接一个地“接收”数字，但您可以跟踪每一步的平均值。

首先，我将在步骤n+1写出均值的公式：

mean[n+1] = mean[n] - (mean[n] - x[n+1]) / (n+1)

初始条件：

mean[0] = x[0]

（索引从零开始）。

第一个方程可以简化为：

mean[n+1] = n * mean[n] / (n+1) + x[n+1]/(n+1)

这个想法是你跟踪平均值，当你“接收”序列中的下一个值时，你会计算出它与当前平均值的偏移量，并将它平均分配给目前看到的 n+1 样本，并相应地调整你的平均值。如果您的数字没有太大的差异，则随着n 变大，您的运行平均值将需要使用新数字进行非常轻微的调整。

显然，即使您在开始时不知道值的总数，此方法也有效。它还有一个额外的优势，即您始终知道当前平均值的值。我能想到的一个缺点是它可能会给一开始看到的数字带来更多的“权重”（不是严格的数学意义上的，而是因为浮点表示）。

最后，如果不够小心，所有此类计算都必然会遇到浮点“错误”。有关浮点计算的一些问题以及如何测试潜在问题，请参阅my answer to another question。

作为测试，我生成了N=100000正态分布的随机数，均值为0，方差为1。然后我通过三种方法计算了它们的均值。

1. sum(numbers) / N，称之为m₁，
2. 我上面的方法，叫m₂,
3. 对数字进行排序，然后用我上面的方法，叫它m₃。

这是我发现的：m₁ − m₂ ∼ -4.6×10^-17, m₁ − m₃ ∼ -3×10^-15, m₂ − m₃ ∼ -3×10 ^-15。因此，如果您的数字已排序，则错误可能对您来说不够小。 （但是请注意，即使是最严重的错误也是 100000 个数字的 10^-15 部分，所以无论如何它可能已经足够好了。）

Answer 8

这里的一些数学解决方案非常好。这是一个简单的技术解决方案。

使用更大的数据类型。这分为两种可能性：

1. 使用高精度浮点库。遇到需要平均十亿个数字的人可能有购买资源或编写 128 位（或更长）浮点库的脑力。

   我了解这里的缺点。它肯定会比使用内在类型慢。如果值的数量增长得太高，您仍然可能会溢出/下溢。亚达亚达。

2. 如果您的值是整数或可以轻松缩放为整数，请将总和保存在整数列表中。当您溢出时，只需添加另一个整数。这本质上是第一个选项的简化实现。下面是 C# 中的一个简单 （未测试） 示例

class BigMeanSet{
    List<uint> list = new List<uint>();

    public double GetAverage(IEnumerable<uint> values){
        list.Clear();
        list.Add(0);

        uint count = 0;

        foreach(uint value in values){
            Add(0, value);
            count++;
        }

        return DivideBy(count);
    }

    void Add(int listIndex, uint value){
        if((list[listIndex] += value) < value){ // then overflow has ocurred
            if(list.Count == listIndex + 1)
                list.Add(0);
            Add(listIndex + 1, 1);
        }
    }

    double DivideBy(uint count){
        const double shift = 4.0 * 1024 * 1024 * 1024;

        double rtn       = 0;
        long   remainder = 0;

        for(int i = list.Count - 1; i >= 0; i--){
            rtn *= shift;
            remainder <<= 32;
            rtn += Math.DivRem(remainder + list[i], count, out remainder);
        }

        rtn += remainder / (double)count;

        return rtn;
    }
}

就像我说的，这是未经测试的——我没有真正想要平均的十亿个值——所以我可能犯了一两个错误，尤其是在 DivideBy 函数中，但它应该展示总体思路。

这应该提供与双精度一样高的精度，并且应该适用于任意数量的 32 位元素，最多 2³² - 1。如果需要更多元素，则 @987654323 @ 变量需要扩展，DivideBy 函数会增加复杂性，但我将把它留给读者练习。

就效率而言，它应该与这里的任何其他技术一样快或更快，因为它只需要遍历列表一次，只执行一次除法运算（好吧，一组），并且完成大部分使用整数。不过，我没有对其进行优化，而且我很确定如果有必要，它还可以稍微快一些。放弃递归函数调用和列表索引将是一个好的开始。再次，为读者做一个练习。该代码旨在易于理解。

如果现在有人比我更有动力，想验证代码的正确性，并解决可能存在的任何问题，请成为我的客人。

---

我现在测试了这段代码，并做了一些小的更正（List<uint> 构造函数调用中缺少一对括号，DivideBy 函数的最终除法中的除数不正确）。

我首先通过 1000 组随机长度（介于 1 和 1000 之间）填充随机整数（介于 0 和 2³² - 1 之间）对其进行测试。这些是我可以通过对它们运行规范均值来轻松快速地验证其准确性的集合。

然后我用 100^* 大系列进行测试，随机长度在 10⁵ 和 10⁹ 之间。这些系列的下限和上限也是随机选择的，受到约束，以便该系列适合 32 位整数的范围。对于任何系列，结果都可以轻松验证为(lowerbound + upperbound) / 2。

_{^{*好吧，这是一个善意的谎言。在成功运行大约 20 或 30 次后，我中止了大系列测试。在我的机器上运行一系列长度为 109 的程序只需要不到一分半钟的时间，所以半小时左右的时间来测试这个程序对我的口味来说已经足够了。}}

有兴趣的朋友，我的测试代码如下：

static IEnumerable<uint> GetSeries(uint lowerbound, uint upperbound){
    for(uint i = lowerbound; i <= upperbound; i++)
        yield return i;
}

static void Test(){
    Console.BufferHeight = 1200;
    Random rnd = new Random();

    for(int i = 0; i < 1000; i++){
        uint[] numbers = new uint[rnd.Next(1, 1000)];
        for(int j = 0; j < numbers.Length; j++)
            numbers[j] = (uint)rnd.Next();

        double sum = 0;
        foreach(uint n in numbers)
            sum += n;

        double avg = sum / numbers.Length;
        double ans = new BigMeanSet().GetAverage(numbers);

        Console.WriteLine("{0}: {1} - {2} = {3}", numbers.Length, avg, ans, avg - ans);

        if(avg != ans)
            Debugger.Break();
    }

    for(int i = 0; i < 100; i++){
        uint length     = (uint)rnd.Next(100000, 1000000001);
        uint lowerbound = (uint)rnd.Next(int.MaxValue - (int)length);
        uint upperbound = lowerbound + length;

        double avg = ((double)lowerbound + upperbound) / 2;
        double ans = new BigMeanSet().GetAverage(GetSeries(lowerbound, upperbound));

        Console.WriteLine("{0}: {1} - {2} = {3}", length, avg, ans, avg - ans);

        if(avg != ans)
            Debugger.Break();
    }
}