如何用Python实现增量式加权均值与协方差的在线计算

你需要在现有代码基础上补充协方差的增量更新逻辑，核心是利用单样本迭代时的均值差外积来更新协方差矩阵，具体实现和说明如下：

修改后完整代码 #

import numpy as np

class OnlineWeightStats:
    def __init__(self, X):
        N, D = X.shape
        self.data_mat = X
        self.mean = np.zeros(D)
        # 初始化协方差矩阵为D×D零矩阵
        self.cov = np.zeros((D, D))
        self.wsum = 1e-7 # 避免初始除以0的平滑项

    def update(self, weights=None):
        if weights is None:
            weights = np.ones(self.data_mat.shape[0])     
        for d, w in zip(self.data_mat,weights):
            old_wsum = self.wsum
            self.wsum += w
            delta = d - self.mean
            # 先更新协方差，再更新均值（delta基于更新前的均值计算）
            self.cov = self.cov * (old_wsum / self.wsum) + np.outer(delta, delta) * (w / self.wsum)
            # 再更新均值
            upcon = w / self.wsum
            self.mean += delta * upcon
    
    def get_unbiased_cov(self):
        """获取无偏估计的协方差矩阵，若不需要无偏可直接访问self.cov"""
        if self.wsum <= 1:
            return self.cov.copy()
        return self.cov * (self.wsum / (self.wsum - 1))

if __name__ == "__main__":
    N = 10000
    D = 2
    X = np.random.rand(N,D)

    # 定义增量统计类
    ows = OnlineWeightStats(X)
    # 第一次更新：默认全1权重
    ows.update()
    # 第二次更新：随机权重
    new_weights = np.random.uniform(0,1,N)
    ows.update(new_weights)

    # 验证结果和离线计算是否一致
    total_weights = np.ones(N) + new_weights
    # 离线计算加权均值
    offline_mean = np.average(X, weights=total_weights, axis=0)
    # 离线计算加权协方差
    diff = X - offline_mean
    offline_cov = np.dot(total_weights * diff.T, diff) / total_weights.sum()
    # 输出差值验证精度
    print("均值误差：", np.max(np.abs(ows.mean - offline_mean)))
    print("协方差误差：", np.max(np.abs(ows.cov - offline_cov)))

逻辑说明 #

• 增量协方差的更新必须在均值更新之前完成，因为计算协方差用到的delta是当前样本和更新前均值的差值
• 代码默认输出的是最大似然估计的协方差，如果需要无偏估计，可以调用get_unbiased_cov()方法
• 你当前的实现是对固定数据集叠加多组权重的场景，如果后续需要支持流式新增样本的增量计算，只需要修改类的逻辑，把单次update的入参改成新样本和对应权重即可，更新逻辑不用改动

内容的提问来源于stack exchange，提问作者ashnair1