目录

Dynamic Time Warping: 从欧式距离到时序对齐

Note

你可以在 这里 找到本篇博客的算法 Demo

最近开始研究如何基于日志定位 LLM 训练过程中遇到的问题,其中涉及到时序数据的分析——迭代训练的时候你可以得到各种时序数据(例如 Loss),看到有的论文里面使用 Dynamic Time Wrapping 分析时序数据中的异常,就记录了这个算法

规定记号如下

  • $\mathbf x_{1:t}$ - 时序数据 $\mathbf x$,长度为 $t$,从 $1$ 开始
  • $\mathbf y_{1:t}$ - 时序数据 $\mathbf y$,长度为 $t$,从 $1$ 开始

想像一下:如果是你,你会如何判断 2 条时序数据 $\mathbf x$ 和 $\mathbf y$ 是否足够相似?

一个很直白的想法是:计算 2 条时序数据之间的距离?

下一个问题是:用什么距离度量?

你可能很快想到欧氏距离,公式也很简单,如下所示

$$ Euclidean(\mathbf x,\mathbf y)=\sqrt{\sum_i(\mathbf x_i-\mathbf y_i)^2} $$

那它可以刻画时序数据之间的距离吗?答案是不行。因为时序数据有一个显著的特点:2 条相似的时序数据在时间这个维度上可能存在 Warping(拉伸/变形)。一个很经典的例子是人说话,不同的人说同一句话,会因为语速不同,在时间这个维度上长度不同

让我们用 $\texttt{sin}$ 函数作为时序数据代表,下面是我 Mock 出来的 2 条时序数据

我们可以一眼看出来这 2 条时序数据是相似的,即便他们在时序上发生了偏移。但欧式距离会强制对齐时序——从公式中也可以看出来,$\mathbf x_i$ 的距离参照是 $\mathbf y_i$,他们都是时序 $i$ 的值,如果我们把这种“强制对齐时序”的关系画出来,会是下面这样

上图的红色虚线就显示了“强制对齐时序”

既然“强制对齐时序”不可取,那我们不强制对齐不久好了?这就是今天要谈到的 Dynamic Time Warping (下称 DTW) 算法,它的核心是 $\mathbf x_i$ 应该找到最佳匹配的 $\mathbf y_j$(这里的 $j$ 不必等于 $i$),找到最佳匹配之后再计算时序数据之间的距离

它是通过如下这些匹配规则实现的

  • 非一对一匹配:每一个 $\mathbf x$ 的数据点都可以和一到多个 $\mathbf y$ 的数据点匹配,反之亦然 ✅️ 能处理时序上的拉伸、压缩关系
  • 连续匹配:每个 $\mathbf x$ 的数据点都需要有匹配 ✅️ 不能忽略其中某一个数据点
  • 边界匹配:$\mathbf x_1$ 和 $\mathbf y_1$ 强制匹配,$\mathbf x_t$ 和 $\mathbf y_t$ 强制匹配 ✅️ 假设 2 条时序数据是相似的,那么他们的开头和结尾肯定应该是匹配的
  • 单调匹配:匹配必须是单调的:如果 $\mathbf x_i$ 和 $\mathbf y_j$ 匹配,那么对于 $\mathbf x_m(m>i)$,它的匹配 $\mathbf y_n$ 必须满足 $n>j$。这么说可能有点绕,但如果我们把节点匹配连成线,这一条规则的意思是你的线不允许相交 ✅️ 时间是单调增加的,不能倒退

DTW 寻找最佳匹配的方式就是——寻找符合上述匹配规则的一条最短路径,用公式表示如下

$$ Dtw(\mathbf x,\mathbf y)=\min_{W}\sum_{(i,j)\in W} d(\mathbf x_i,\mathbf y_j) $$

这里的 $W$ 就是一条可能的路径,可以看成是一个字典,表示 $\mathbf x$ 的每一个数据点应该和哪些 $\mathbf y$ 里的数据点匹配,$d$ 可以使用欧式距离

对于每一个给定的时序匹配关系 $W$,我们都可以计算该时序匹配关系下时序数据的距离,这里的距离函数 $d$ 可以用前面提到的欧式距离

现在我们清楚了 DTW 算法的原理,下一个问题就是——如何实现?

在谈论实现之前,我们可以换另外一种视角看 DTW 要解决的时序数据匹配问题。我们分别用行和列表示 2 条时序数据,那么 DTW 要寻找的最佳匹配恰好是这个网格上的从左上角走到右下角的最短路径

$\mathbf y_1$ $\mathbf y_2$ $\mathbf y_3$
$\mathbf x_1$ $d(\mathbf x_1, \mathbf y_1)$ $d(\mathbf x_1, \mathbf y_2)$ $d(\mathbf x_1, \mathbf y_3)$
$\mathbf x_2$ $d(\mathbf x_2, \mathbf y_1)$ $d(\mathbf x_2, \mathbf y_2)$ $d(\mathbf x_2, \mathbf y_3)$
$\mathbf x_3$ $d(\mathbf x_3, \mathbf y_1)$ $d(\mathbf x_3, \mathbf y_2)$ $d(\mathbf x_3, \mathbf y_3)$

为什么这么说?

  • 非一对一匹配:在同一行内你可以选择从左到右➡️走
  • 连续匹配:你可以从上到下⬇️走,或者从左上方走到右下方↘️
  • 边界匹配:起点是左上角,重点是右下角
  • 单调匹配:你不能走“回头路”,不能向左或者向左下方走

我们不难发现,这是一个动态规划问题,在每个状态(坐标) (x, y) 的位置,我们可以选择

  • 向右走:(x, y) -> (x, y + 1)
  • 向下走:(x, y) -> (x + 1, y)
  • 向右下角走:(x, y) -> (x + 1, y + 1)

下面是示例代码,该代码返回 DTW 的最佳匹配下的最短距离和最佳匹配

def dtw(series_a: np.ndarray, series_b: np.ndarray) -> dict[int, list[int]]:
    m, n = len(series_a), len(series_b)
    dp = [[float("inf") for _ in range(n + 1)] for _ in range (m + 1)]
    dp[0][0] = 0
    
    for i in range(1, m + 1):
        for j in range(1, n + 1):
            dp[i][j] = min(
                dp[i - 1][j - 1],
                dp[i - 1][j],
                dp[i][j - 1],
            ) + abs(series_a[i - 1] - series_b[j - 1])

    # backtrace
    matches: dict[int, list[int]] = {}
    i, j = m, n
    while i > 0 and j > 0:
        matches.setdefault(i - 1, []).append(j - 1)
        candidates = [
            (dp[i - 1][j - 1], i - 1, j - 1),
            (dp[i][j - 1], i, j - 1),
            (dp[i - 1][j], i - 1, j),
        ]
        _, i, j = min(candidates, key=lambda t: t[0])

    return dp[-1][-1], matches

那么 DTW 最佳匹配是怎么样的?可以从下图中看出来

不难看出来 DTW 找的匹配确实很符合时序关系 :)

也可以用量化的数据比较一下 2 种方式计算出来的距离

  • 纯欧氏距离:7.04
  • DTW 寻找最佳匹配 + 欧式距离:5.13

DTW 用于找到 2 条时序数据的最佳匹配。在最佳匹配下,时序数据不会被强制对齐,而是会根据时序数据的特点找到符合的匹配,这样计算出来的时序数据距离更靠谱