如何实现Python中低耗时的DTW算法以完成大规模时间序列聚类

我有1977位客户的时间序列数据，每个序列包含17544个数据点（两年的小时级数据），需要将客户划分到对应聚类群组。以下是我的实现代码（list_of_lists为存储各客户时间序列的列表）：

from dtw import dtw
import numpy as np

# 定义自定义距离函数
def my_dist(x, y):
   return np.abs(x - y) 

# 计算两两DTW距离
distances = np.zeros((len(list_of_lists), len(list_of_lists)))
for i in range(len(list_of_lists)):
  for j in range(i+1, len(list_of_lists)):
      x = list_of_lists[i]
      y = list_of_lists[j]
      distance, *rest = dtw(x, y, dist=my_dist)
      distances[i,j] = distance
      distances[j,i] = distance

from sklearn.cluster import AgglomerativeClustering
# 基于距离矩阵执行层次聚类
clustering = AgglomerativeClustering(n_clusters=cluster_count, metric=None, 
            linkage='average')
labels = clustering.fit_predict(distances)
print(labels)

但当前代码计算耗时过长，以下是几种能大幅减少计算时间的优化方案：

1. 替换为更快的DTW近似实现 #

原生DTW是精确算法，时间复杂度为O(n*m)（n、m为序列长度），对于17544长度的序列，单次计算成本极高。fastdtw是DTW的近似实现，时间复杂度降至O(n+m)，速度提升数十倍，且精度损失在可接受范围内。

安装依赖：

pip install fastdtw

修改代码：

from fastdtw import fastdtw
import numpy as np

# 直接用numpy的L1距离（比自定义Python函数快）
def my_dist(x, y):
    return np.linalg.norm(x - y, ord=1)

# 计算距离矩阵
n_samples = len(list_of_lists)
distances = np.zeros((n_samples, n_samples))
for i in range(n_samples):
    for j in range(i+1, n_samples):
        distance, _ = fastdtw(list_of_lists[i], list_of_lists[j], dist=my_dist)
        distances[i,j] = distance
        distances[j,i] = distance

2. 对时间序列做降维处理 #

17544个小时级数据包含大量冗余信息，可通过聚合或特征提取缩短序列长度：

• 按时间周期聚合：比如将每24小时的小时数据聚合为日均值/最大值/最小值，序列长度从17544压缩到730（两年共730天）；
• 提取周期特征：针对小时级数据，提取每周/每日的模式特征（比如每天各小时的均值、每周各天的均值）；
• PCA降维：对序列做PCA，保留95%以上的方差，将高维序列压缩到低维空间。

示例：按天聚合小时数据

# 假设list_of_lists中每个元素是形状为(17544,)的numpy数组
aggregated_series = []
for series in list_of_lists:
    # 重塑为(730, 24)，按天分组
    daily_groups = series.reshape(-1, 24)
    # 取每天的均值作为聚合后的数据点
    daily_mean = daily_groups.mean(axis=1)
    aggregated_series.append(daily_mean)
# 之后用aggregated_series代替原list_of_lists计算距离

3. 并行化距离矩阵计算 #

原代码用单线程循环计算两两距离，可通过多进程并行加速，利用CPU多核资源。使用joblib实现并行：

from joblib import Parallel, delayed
import numpy as np
from fastdtw import fastdtw

def compute_pairwise_dist(i, j, data):
    distance, _ = fastdtw(data[i], data[j], dist=lambda x,y: np.linalg.norm(x-y, ord=1))
    return (i, j, distance)

n_samples = len(aggregated_series)
# 生成所有i<j的索引对
index_pairs = [(i,j) for i in range(n_samples) for j in range(i+1, n_samples)]

# 并行计算，n_jobs设为CPU核心数
results = Parallel(n_jobs=-1)(delayed(compute_pairwise_dist)(i,j,aggregated_series) for i,j in index_pairs)

# 填充距离矩阵
distances = np.zeros((n_samples, n_samples))
for i,j,d in results:
    distances[i,j] = d
    distances[j,i] = d

4. 更换聚类算法，避免全量距离矩阵计算 #

层次聚类需要预先计算O(n²)的全量距离矩阵，对于1977个样本，这需要近400万次DTW计算。可更换为无需全量距离矩阵的算法：

• HDBSCAN：基于密度的聚类，支持自定义距离函数，可通过近似最近邻减少计算量；
• DBSCAN：配合BallTree/KDTree加速最近邻搜索；
• k-medoids（PAM）：用sklearn_extra.cluster.KMedoids，可自定义距离，但需注意样本量。

示例：用HDBSCAN结合fastdtw

import hdbscan
from fastdtw import fastdtw

# 定义距离函数供HDBSCAN使用
def dtw_distance(x, y):
    distance, _ = fastdtw(x, y, dist=lambda a,b: np.linalg.norm(a-b, ord=1))
    return distance

# HDBSCAN会自动处理距离计算，无需预先生成全量矩阵
clusterer = hdbscan.HDBSCAN(metric=dtw_distance, min_cluster_size=5)
labels = clusterer.fit_predict(aggregated_series)

5. 使用内置优化的距离函数 #

原代码自定义的my_dist是Python层面的函数，调用开销大。直接使用numpy的内置距离计算，或fastdtw支持的内置距离（比如'manhattan'），这些实现经过C级优化，速度更快：

# 无需自定义函数，直接传入'manhattan'给fastdtw
distance, _ = fastdtw(x, y, dist='manhattan')

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