c# - 快速计算输入数字的最小值、最大值和平均值

Question

程序每秒接收大约 50,000 个号码。

在任何给定时刻，我都需要计算在最后一秒（关于给定时刻）到达的值（数字）的最小值、最大值和平均值。

有没有办法在不使用数组或列表（缓冲区）存储到达的数字和计算结果的情况下做到这一点？

如果我需要使用缓冲区，那么实现这一目标的有效方法是什么？

（请注意，缓冲区中的数字也必须不时有效地删除）

Answer 1

这是一种算法，在某些情况下可以提高效率：

1. 当事件进入时，将它们完全缓冲，并计算一个运行的sum, count, min, max（微不足道的）。

2. 当请求average,min或max时，从缓冲区的后面循环并开始删除早于一秒的值。减去sum和count当你去。

   • 如果这些值都在上面，min你可以保留你的min. 如果值低于max，您可以保留您的max. 在这种情况下，您已经有效地更新了average、min和。max

   • 如果值低于min或高于max，您将需要遍历数组的其余部分并重新计算它。

3. 每秒执行一次第二步左右，这样缓冲区就不会太满。该代码也可以在每个缓冲区插入上执行，或者在任何有意义的地方执行。

这种工作的最佳结构是循环缓冲区，以避免内存分配和 GC 妨碍。它应该足够大以涵盖每秒消息大小的最坏情况。

更新

根据使用场景，另一件事是运行上面的算法，但以 10 x 100ms 块而不是 1 x 1000ms 块运行。也就是说，在这 10 个块上保持运行的 min、max、sum 和 count。然后，当您遇到“无效”情况时，您通常只需要查看最新的 100ms 数据或快速浏览其他 9 个块的最小值和最大值。

---

@ja72 提供了一个好主意，如果它们无效，可以节省查找最小值和最大值：

x_max 不是保留最小值/最大值 x_min，而是使用 i_min 和 i_max 保留它们在 x[i] 数组中的位置索引。然后找到它们有时可能是微不足道的，但是当考虑的最后一个值包含最小值和最大值时，需要扫描整个列表以建立新的限制。

---

Sam Holder 在评论中有另一个好主意 - 保持一个始终排序的并行数组，这让您可以将数字从顶部或底部剔除，以便更轻松地找到新的最小值和最大值。但是，此处的插入速度会受到一点影响（需要保持有序）。

---

最终，正确的选择将取决于程序的使用特性。读取值的频率与插入的频率？

Answer 2

使用具有时间戳和数据的每个元素的循环缓冲区，每秒元素的最大数量作为循环缓冲区的大小。

当每个元素都插入缓冲区头时，检查缓冲区另一侧的过期时间，删除该元素。

如果删除的元素是最小值或最大值，则必须计算新的最小值/最大值。如果不是，您将根据新来者更新最小/最大值。

对于平均值，保留总数，保留计数，然后除法。

Answer 3

您不能将您的号码及其到达时间以及队列中当前的最大值和最小值（可能需要将值的数量保持在相同的最小值/最大值）和所有的总值保持一个队列队列中的数字和元素计数。

然后当一个数字到达时将其添加到队列中并调整最小/最大值/值和计数。然后查看队列的另一端，将最后一个数字到达后1秒内没有的元素全部移除，再次调整max/min/count/total值。

然后你不需要立即计算任何东西，只需返回预先计算的东西（即读取最小/最大值或总计/计数的当前值）

正如@yaman 指出的那样，您不能只保留最小值和最大值，因为当一个被删除时，您可能不知道新的。在这种情况下，我可能只保留列表中所有数字的第二份副本，而不是按到达时间排序，而是按值排序。然后，您也只需从该列表中添加和删除每个数字，因此您将始终知道最大值和最小值。这使您不必扫描缓冲区中的所有元素以找到新的最大值/最小值，但代价是保留 2 个副本，但此列表的更新应该很便宜，因为它已经订购了。

Answer 4

如果最后一个值的N平均值x[0]..x[N-1]是m_1（x[0]是最新值，并且x[N-1]是考虑的最后一个值），那么m_2将所有值推回一个索引并添加该值的值的平均值x是

 m_2 = m_1+(x-x[N-1])/N;
 for(i=N-1;i>0;i--) { x[i]=x[i-1]; }
 x[0] = x;

与其保持最小值/最大值x_min，不如使用和x_max保持它们在x[i]数组中的位置索引。然后找到它们有时可能是微不足道的，但是当考虑的最后一个值包含最小值和最大值时，需要扫描整个列表以建立新的限制。i_mini_max

Answer 5

有一种有效的方法可以跟踪给定时间窗口内的最小值（或最大值），而通常不必存储在该窗口内到达的所有数字。（但是，最坏的情况仍然需要存储所有数字，因此您需要为所有数字保留空间，或者接受您有时可能会得到不正确的结果。）

诀窍是只存储以下值：

1. 已在时间窗口内到达，并且
2. 小于（或大于）任何以后的值。

实现这一点的合适数据结构是一个简单的循环缓冲区，用于存储值及其到达时间。您需要在缓冲区中维护两个索引。以下是该算法的简单英文描述：

启动时：

• 分配一个N元素的值缓冲区val和一个对应的N元素的时间戳缓冲区time。
• 让imax= 0（或 0 和N -1 之间的任何其他值）并且让inext= imax。这表明缓冲区当前是空的。

在 time 收到新值 new 时 t：

• 当imax≠inext和time[imax]在区间外时，加imax一（模N）。
• 当imax≠inext和val[inext-1]≥new时，减inext一（模N）。
• 让val[inext]=new和time[inext]= t。
• 如果inext≠ imax-1，则加inext一（模N）；否则适当地处理“缓冲区已满”的情况（例如分配一个更大的缓冲区，抛出一个异常，或者只是忽略它并接受最后一个值没有正确记录）。

当请求最小值时：

• 当imax≠inext和time[imax]在区间外时，加imax一（模N）。
• 如果imax≠ inext，则返回val[imax]；否则返回一个错误，表明在时间间隔内没有收到任何值。

如果接收到的值是独立且同分布的（并且作为泊松过程到达），我相信可以证明在任何给定时间存储在列表中的值的平均数是 ln( n +1)，其中n是在时间间隔内接收到的值的平均数。对于n = 50,000，ln( n +1) ≈ 10.82。但是，应该记住，这只是平均值，有时可能需要多几倍的空间。

---

对于平均而言，不幸的是，同样的技巧不起作用。如果可能的话，您可以切换到指数移动平均线，它可以使用非常小的空间轻松跟踪（平均只有一个数字和一个指示上次更新时间的时间戳）。

如果这是不可能的，但您愿意接受平均值的少量平滑，您可以计算一个平均值，例如，每毫秒。这样，每当请求最后一秒的平均值时，您可以取最后 1001 毫秒平均值的平均值，根据间隔内的毫秒数来衡量最旧和最新的平均值：

启动时：

• 让interval是要平均的时间间隔的长度，让n是子间隔的数量。
• 令dt =区间/ n。
• 分配一个n +1 元素sum的值缓冲区和一个n +1 元素cnt的非负整数缓冲区，并用零填充两者。
• 让prev有任何价值。（这并不重要。）

在 time 收到新值 new 时 t：

• 让i= floor( t/ dt ) mod ( n +1)。
• 如果i≠ prev：
  • sum[i]从total和cnt[i]中减去count。
  • 让sum[i]= 0, cnt[i]= 0 和让prev= i。
• 加new一并sum[i]递增cnt[i]。
• 加new一并total递增count。

当在 time 请求平均值时 t：

• 让i= floor( t/ dt ) mod ( n +1)。
• 如果i≠ prev：
  • sum[i]从total和cnt[i]中减去count。
  • 让sum[i]= 0、cnt[i]= 0 和让prev= i。
• 令j= ( i- n ) mod ( n +1) = ( i+1) mod ( n +1)。
• 让w= frac( t/ dt ) = ( t/ dt ) − floor( t/ dt )。
• 返回 ( total- w× sum[j]) / ( count- w× cnt[j])。

Answer 6

@DanRedux 是正确的；您每次都需要计算它们，因为您的输入正在改变。现在，您可能更喜欢按需或预先（即，当您获得新批次时）计算这些数字，具体取决于需要结果的频率。

例如，如果您的平均用例每约 30 秒轮询一次这些统计信息，那么我可能会按需计算它们并缓存结果，直到有新批次进来。不过，这实际上取决于您的使用场景。

至于如何存储它们，你真的没有选择，是吗？您需要内存中所有 50,000 个数字的空间。所以......你需要一块足够大的内存来容纳它们。为了避免每次有新序列进入时不断分配 2KB，您最好声明一个足够大的数组以容纳可能的最大数据集并重用它。同样，这取决于您的要求，即，您知道您的最大可能数据集是多少吗？随着时间的推移，每秒钟分配一块新的内存是否会导致应用程序出现问题？

Answer 7

有没有办法在不使用数组或列表（缓冲区）存储到达的数字和计算结果的情况下做到这一点？

不，正如您所说，如果不存储信息，这可能是不可能的。不过，您可以稍微调整要求以摆脱对缓冲区的需求。

如果我需要使用缓冲区，那么实现这一目标的有效方法是什么？

您需要为此使用队列。

添加项目时，如果它是新的最大值或最小值，则相应地调整这些变量。您可以通过此处的公式逐步调整平均值。只需将新值减去平均值除以集合中的新项目数（即队列大小加一），然后将其添加到旧平均值。

然后你或多或少会有这样的东西：

while(queue.Peek < oneSecondAgo)
{
  oldItem = queue.Peek
  queue.Dequeue();
  if(oldItem == min) //recalculate min
  if(oldItem == max) //recalculate max
  mean += SubtractValueFromMean(oldItem.Value, queue.Count);
}

要从平均值中删除值，您应该能够使用相同的公式进行添加，但使用值的负数而不是正数......我认为。一个更好的数学家可能需要在这里帮助你。

Answer 8

如果数字一个接一个地出现，则使用秒表和 while 循环在一秒钟内一个接一个地获取每个数字，并计算最小值、最大值和平均值。

double min = double.MaxValue;
double max = double.MinValue;
double sum = 0;
int count = 0;
double avg;
StopWatch sw = new StopWatch();
sw.Start();
while(sw.Elapsed.TotalSeconds <= 1)
{
   // Get the next number in the stream of numbers
   double d = GetNextNumber();

   // Calculate min
   if(d < min) min = d;
   // Calculate max
   if(d > max) max = d;

   // Calculate avg = sum/ count
   sum += d;
   count++;
}

avg = sum/count;

然后返回最小值、最大值和平均值。

Answer 9

Sadly, there isn't. The reason why it's not possible is because you need to only account for them that are a second old, which means you have to re-calculate the result every time, which means HUGE loops.

If you wanted to calculate the last 40,000 numbers, or all of them, it would be easier, but because it's time-based, you have to loop through the entire list every time.

Answer 10

不保留缓冲区或队列中的数字是不可能的。

原因很简单：当最大值过期（超出 1 秒窗口）时，新的最大值是在最后一秒内到达的其他数字，因此您需要记录可能成为新的最大值。

需要平均值意味着所有值在它们过期时都会产生影响，并且在它一秒钟之前不会丢弃任何东西。

Sam Holder 关于使用队列的建议是一个很好的建议，尽管您可能需要一个专门的队列来同时将您的列表保持在两个顺序中：接收数字的顺序（到达时间），并从最大到最小排序.

使用带有两个下一个和两个前一个指针的单个节点对象（一对是临时的，另一对是大小的）可以同时从两个列表中删除元素，当一个元素从临时列表中过期时，您可以访问大小列表的指针，因为它们在同一个节点对象中。

可以通过保持运行总数和运行计数来保持平均值，在删除元素时减去元素，并在创建元素时添加它们，因此不必每次都遍历整个列表来计算平均值。

正如 btilly 在他们对 Sam Holder 帖子的评论中所建议的那样，使用最大堆和最小堆比使用列表更有效，我们将再次需要使用单个节点，同时为堆和列表使用指针，所以我们不必搜索元素来删除它们，并且可能需要花一些时间考虑如何正确删除不在堆顶部的元素，同时保持 O(log n) 插入和删除的保证。

Answer 11

平均而言，需要考虑 3 种情况：

1. 你的数字是整数。保持一个运行的总数和计数，将新值添加到总数中，从总数中减去旧值，并根据需要除以计数。这很简单，因为您不必担心精度损失。
2. 您的数字是浮点数，您需要 0 的精度损失：您必须遍历整个一秒列表来计算平均值
3. 您的数字是浮点数，您可以忍受一些精度损失：按整数平均值操作，每 1000 个值左右进行一次完整的重新计算。

对于最小值和最大值（仅与上述 #1 和 #3 相关）：

• 将值保存在按值索引的陷阱中。
• 还要将值保存在按时间排序的双向链表中。保存列表的开头和结尾。
• 从列表的开头删除，并添加到列表的末尾。
• 对于每个新值：将其添加到时间链表的开头。根据需要从时间链表的末尾删除值。

当您在链表中添加和删除值时，对 treap 执行相应的操作。要从 treap 中获取最小值和最大值，只需在 log(n) 时间内执行 find_minimum 和 find_maximum 操作。当您在恒定时间内从链表的右端删除东西时，也要在 log(n) 时间内从treap 中删除它们。

Treaps 可以在 log(n) 时间内找到它们的最小值，在 log(n) 时间内找到它们的最大值，并在 log(n) 时间内找到任意值。一般来说，访问数据所需的不同方式越多，像treap 这样全面的数据结构就越好。