如何用Numpy高效计算时序高维向量与前置均值的余弦滚动距离？

我有一个向量时间序列：Y = [v1, v2, ..., vn]。在每个时刻t，需要计算向量v_t与t时刻之前所有向量的平均值之间的余弦距离。例如t=3时，要计算v3与(v1+v2)/2之间的余弦距离。

我已经编写了实现该功能的脚本，但想了解是否可通过Numpy的卷积功能或类似方法提升计算速度？

原实现代码：

import numpy as np
from scipy.spatial.distance import cosine

np.random.seed(10)

# 生成`T`个维度为`vector_dim`的向量
# 注意：实际场景中，向量是一个非常大的列向量！
T = 3
vector_dim = 2
y = [np.random.rand(1, vector_dim)[0] for t in range(T)]

def moving_distance(v):
    moving_dists = []
    for t in range(len(v)):
        if t == 0: 
            pass
        else:
            # 生成t时刻之前所有向量的移动平均值
            prior_vals = v[:t]
            m_avg = np.add.reduce(prior_vals) / len(prior_vals) 
            # 计算该移动平均值与t时刻向量的距离
            moving_dists.append(cosine(m_avg, v[t]))
    return moving_dists

d = moving_distance(y)

针对该数据集，预期返回结果为：[0.3337342770170698, 0.0029993196890111262]

优化方案：用累积和实现向量化计算 #

原代码的循环遍历会重复计算前t个向量的和，效率较低。我们可以用Numpy的cumsum计算累积和，避免重复运算，同时手动实现余弦距离的向量化计算，大幅提升速度，尤其是在向量维度大、时间步多的场景下。

优化后的代码（半向量化）：

import numpy as np

np.random.seed(10)

T = 3
vector_dim = 2
y = [np.random.rand(1, vector_dim)[0] for t in range(T)]
y_arr = np.array(y)  # 转换为(N, D)的Numpy数组

# 计算累积和：cum_sum[i] 是前i+1个向量的和（从索引0到i）
cum_sum = np.cumsum(y_arr, axis=0)

# 初始化结果数组
moving_dists = np.zeros(T-1)

for t in range(1, T):
    # 前t个向量的和是cum_sum[t-1]，平均值为 sum / t
    m_avg = cum_sum[t-1] / t
    # 计算余弦距离：1 - (a·b)/(||a|| * ||b||)
    dot_product = np.dot(m_avg, y_arr[t])
    norm_avg = np.linalg.norm(m_avg)
    norm_vt = np.linalg.norm(y_arr[t])
    moving_dists[t-1] = 1 - (dot_product / (norm_avg * norm_vt))

print(moving_dists)
# 输出：[0.33373428 0.00299932]，与预期结果一致

完全向量化实现（无循环） #

如果时间步T非常大，可以把整个计算过程完全向量化，彻底去掉循环，进一步提升效率：

import numpy as np

np.random.seed(10)

T = 3
vector_dim = 2
y_arr = np.random.rand(T, vector_dim)

cum_sum = np.cumsum(y_arr, axis=0)
# 前t个向量的和：取cum_sum的前T-1项（对应t=1到T-1时的前t个向量和）
prior_sums = cum_sum[:-1]
# 每个时刻的除数t：1到T-1，转换为(T-1,1)形状方便广播
divisors = np.arange(1, T)[:, np.newaxis]
m_avgs = prior_sums / divisors

# 计算点积、范数，最终得到余弦距离
dot_products = np.sum(m_avgs * y_arr[1:], axis=1)
norm_avgs = np.linalg.norm(m_avgs, axis=1)
norm_vts = np.linalg.norm(y_arr[1:], axis=1)
moving_dists = 1 - (dot_products / (norm_avgs * norm_vts))

print(moving_dists)
# 输出：[0.33373428 0.00299932]

为什么不用卷积？ #

卷积主要用于固定窗口的滑动线性组合计算，但这里的需求是累积平均（窗口从第一个元素到当前元素的前一个），用累积和cumsum比卷积更直接高效，不需要构造额外的卷积核，计算逻辑更清晰，速度也更快。

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