DTW——动态时间调整

优质博文：

• https://www.cnblogs.com/xingshansi/p/6924911.html
• https://blog.csdn.net/raym0ndkwan/article/details/45614813
• https://github.com/wannesm/dtaidistance
• https://www.cnblogs.com/ningjing213/p/10502519.html

基本原理

一目了然

优质代码

这段代码条例清晰、言简意赅，虽然效率有待提升，但仍然值得学习

def dtw_distance(ts_a, ts_b, d=lambda x,y: abs(x-y), mww=np.inf):
    """Computes dtw distance between two time series
    
    Args:
        ts_a: time series a
        ts_b: time series b
        d: distance function
        mww: max warping window, int, optional (default = infinity)
        
    Returns:
        dtw distance
    """
    
    # Create cost matrix via broadcasting with large int
    ts_a, ts_b = np.array(ts_a), np.array(ts_b)
    M, N = len(ts_a), len(ts_b)
    cost = np.ones((M, N))

    # Initialize the first row and column
    cost[0, 0] = d(ts_a[0], ts_b[0])
    for i in range(1, M):
        cost[i, 0] = cost[i-1, 0] + d(ts_a[i], ts_b[0])

    for j in range(1, N):
        cost[0, j] = cost[0, j-1] + d(ts_a[0], ts_b[j])

    # Populate rest of cost matrix within window
    for i in range(1, M):
        for j in range(max(1, i - mww), min(N, i + mww)):
            choices = cost[i-1, j-1], cost[i, j-1], cost[i-1, j]
            cost[i, j] = min(choices) + d(ts_a[i], ts_b[j])

    # Return DTW distance given window 
    return cost[-1, -1], _traceback(cost)

def _traceback(D):
    i, j = array(D.shape) - 1
    path = [(i,j)]
    while ((i > 0) or (j > 0)):
        back = argmin((D[i-1, j-1], D[i, j-1], D[i-1, j]))
        if (back == 0):
            i -= 1
            j -= 1
        elif (back == 1):
            j -= 1
        else:
            i -= 1
        path.append((i,j))
    return list(reversed(path))

结果可视化

比较信号

动态调整结果

画图代码

import matplotlib.pyplot as plt
import matplotlib as mpl

def plot_warping(s1, s2, path):
    """Plot the optimal warping between to sequences.
    :param s1: From sequence.
    :param s2: To sequence.
    :param path: Optimal warping path.
    """
    
    fig, ax = plt.subplots(nrows=2, ncols=1, sharex=True, sharey=True)
    ax[0].plot(s1)
    ax[1].plot(s2)
    transFigure = fig.transFigure.inverted()
    lines = []
    line_options = {'linewidth': 0.5, 'color': 'orange', 'alpha': 0.8}
    for r_c, c_c in path:
        if r_c < 0 or c_c < 0:
            continue
        coord1 = transFigure.transform(ax[0].transData.transform([r_c, s1[r_c]]))
        coord2 = transFigure.transform(ax[1].transData.transform([c_c, s2[c_c]]))
        lines.append(mpl.lines.Line2D((coord1[0], coord2[0]), (coord1[1], coord2[1]),
                                      transform=fig.transFigure, **line_options))
    fig.lines = lines
    
    return fig, ax


s1 = [np.sin(i) for i in range(20)]
s2 = [np.sin(i+0.1+np.pi) for i in range(20)]
plt.plot(s1)
plt.plot(s2)

dis, path = dtw_distance(s1, s2)
plot_warping(s1, s2, path)

其它例子