功能说明
监控、批处理和日志分析常常需要在不下发全量数据的情况下计算均值与标准差。两遍算法要先求均值再求平方和，无法处理流式输入；而直接累加平方和再减去均值平方的公式，在数值很大且方差很小时会发生灾难性抵消，甚至得到负数方差。Welford 算法用增量方式维护均值与二阶中心矩，只需一次遍历，数值稳定性明显更好。
本例还提供 merge，把两个独立统计结果合并。分片计算后再汇总时不需要保留原始样本，适合并行处理和分布式聚合。
关键代码
add 先增加计数，计算当前值与旧均值的差 delta，再用 delta 除以新计数更新均值。随后用 delta 乘以新值与新均值之差累加到 m2。这个顺序保证每一步都围绕当前均值计算，避免大平方项相减。variance 在样本数小于二时返回零，否则用 m2 除以 n 减一得到样本方差；如需总体方差可改为除以 n。stddev 对方差开方。merge 使用并行方差合并公式，先按两组样本量加权更新均值，再把组内 m2 与组间差异项相加。
使用注意
该实现使用样本方差，除数是 n 减一；总体方差应改为 n。输入中若含 NaN 或无穷大会污染后续所有统计值，应在添加前校验。count 为一时标准差返回零只是约定，不能据此判断数据没有波动。合并只适用于同一总体下的两组数据，且两组都使用相同的方差定义。浮点误差仍然存在，极端量级数据可考虑用 decimal 或 math.fsum 辅助。
验证方式
运行脚本，样本数量为八，均值应为五，样本标准差约为二点一三八一。合并前三个与后五个数据后，均值与标准差应与一次性计算完全一致，输出 merged: 5.0 2.1381。可把数据放大一亿倍后再比较两种算法的差异，Welford 不会出现负方差。
代码示例
from dataclasses import dataclass
@dataclass
class Welford:
count: int = 0
mean: float = 0.0
m2: float = 0.0
def add(self, value: float) -> None:
self.count += 1
delta = value - self.mean
self.mean += delta / self.count
self.m2 += delta * (value - self.mean)
@property
def variance(self) -> float:
if self.count < 2:
return 0.0
return self.m2 / (self.count - 1)
@property
def stddev(self) -> float:
return self.variance ** 0.5
def merge(self, other: "Welford") -> None:
if other.count == 0:
return
total = self.count + other.count
delta = other.mean - self.mean
self.mean += delta * other.count / total
self.m2 += other.m2 + delta * delta * self.count * other.count / total
self.count = total
def summarize(values: list[float]) -> dict[str, float | int]:
stats = Welford()
for value in values:
stats.add(value)
return {
"count": stats.count,
"mean": round(stats.mean, 4),
"stddev": round(stats.stddev, 4),
}
if __name__ == "__main__":
sample = [2.0, 4.0, 4.0, 4.0, 5.0, 5.0, 7.0, 9.0]
print(summarize(sample))
left, right = Welford(), Welford()
for value in sample[:3]:
left.add(value)
for value in sample[3:]:
right.add(value)
left.merge(right)
print("merged:", round(left.mean, 4), round(left.stddev, 4))