[PYTHON] Comment comparer des données de séries chronologiques

emballer

Objectif: mesurer la similitude des différentes données de séries chronologiques
Mots clés: déformation temporelle dynamique (DTW), déformation temporelle dynamique dérivée (DDTW)
Référence 1: DTW dérivée ~ Méthode de calcul de la similarité du modèle de série chronologique ~ par Tech Blog
Référence 2: [DTW (Dynamic Time Warping) Dynamic Time Warping par S-Analysis](https://data-analysis-stats.jp/%E6%A9%9F%E6%A2%B0%E5% AD% A6% E7% BF% 92 / dtwdynamic-time-warping% E5% 8B% 95% E7% 9A% 84% E6% 99% 82% E9% 96% 93% E4% BC% B8% E7% B8% AE% E6% B3% 95 /)
Référence 3: Implémentation de DTW (Dynamic Time Stretching Method) par ksknw
Référence 4: Comment utiliser fastdtw

DTW

Comment trouver DTW

Créez une matrice (de coût) qui calcule la distance entre chaque point des deux séries chronologiques (cette fois, il s'agit d'une valeur absolue, en fonction de la dimension des données)

\begin{align}
time\ series\ T&:t_0,t_1,...,t_N,\\
time\ series\ S&:s_0,s_1,...,s_M,\\
\boldsymbol{W}&:=(\delta(i,j)) ,\\
where\ \delta(t_i,s_j)&:=|t_i-s_j|,i \in \{0,1,...,N\},j\in\{0,1,...,M\} 
\end{align}

Considérons un chemin qui passe de (0,0) à (N, M) sur cette matrice et satisfait l'augmentation monotone.

\begin{align}
path\ \tau :\tau(0)=(0,0),\tau(1),...,\tau(p)=(p_i,p_j),...,\tau(K),\\
where\ p_i \in \{0,1,...,N\},p_j\in\{0,1,...,M\}.
\end{align}

En particulier, considérons quelque chose qui satisfait l'augmentation monotone avec ce $ chemin \ \ tau $.

```math
\tau(p) \leq \tau(q) \Longleftrightarrow^{def} p_i\leq q_i\ or\ p_j\leq q_j.\\
if\ p\leq q,then\ \tau(p) \leq \tau(q).
```

Soit le coût DTW la plus petite valeur de la somme des éléments de la matrice $ \ boldsymbol {W} $ par laquelle ce $ chemin $ passe.
```
\begin{align}
cost(\tau)&:=\sum_{p}\delta(p_i,p_j)\\
Dist(T,S)&:=min_{\tau}cost(\tau)
\end{align}
```

La figure ci-dessous est tirée de Derivative Dynamic Time Warping by Eamonn

C'est différent de la formule officielle, mais ça ressemble à ça. Code de test: Référence 3

#Préparation
pip install fastdtw # <=pip est en fait juste celui-ci
pip install numpy
pip install scipy

##code##
import numpy as np
from scipy.spatial.distance import euclidean
from fastdtw import fastdtw
A = np.sin(np.array(range(T)) / 10)
B = np.sin((np.array(range(T)) / 10 + t * np.pi))
C = np.zeros((T))
distance, path = fastdtw(A, B, dist=euclidean)
print(distance)
plt.plot(A)
plt.plot(B)
for i, j in path:
   plt.plot([i, j], [A[i], B[j]],color='gray', linestyle='dotted')
plt.legend(["sin(θ)", "sin(θ+150*pi)"], fontsize=10, loc=2)
plt.show()

##résultat##
sin(θ)-sin(θ+150*pi): 0.6639470476737607
sin(θ)-constant: 0.5150026855435942
DTW(sin(θ), sin(θ+150*pi)): 16.720461687388624
DTW(sin(θ), constant): 97.26964355198544

Avantages de DTW

La similitude peut être obtenue même si les longueurs et les cycles des séries chronologiques sont différents.

Inconvénients de DTW

Un alignement non intuitif est effectué pour les données de séries temporelles qui ont des parties accélérées et décélérées localement sur l'axe des temps: Référence: Dérivée Dynamic Time Warping par Eamonn, etc. Par exemple, si l'un d'entre eux devient s haut, DTW ignore la forme des données et s'aligne, donc la cause est que le point de données de s hauteur est associé au point de données avant s hauteur.

DDTW

Recevez DTW et mesurez le degré de similitude en ajoutant des informations qui capturent la forme. En fait, après avoir traité les données de séries temporelles $ time \ series \ T: t_0, t_1, ..., t_N $ comme suit, DTW:

T^*:t^*_0,t^*_1,...,t^*_N,\\
where\ t^*_i :=  \frac{t_i-t_{i-1}+\frac{t_{i+1}-t_{i-1}}{2}}{2}=  \frac{t_i-t_{i-1}+\frac{(t_{i+1}-t_i)+(t_i-t_{i-1})}{2}}{2}

Ce processus est la moyenne du différentiel gauche $ t_i-t_ {i-1} $ et du différentiel droit $ t_ {i + 1} -t_i $ et la moyenne du différentiel gauche pour le point d'intérêt $ t_i $. .. DDTW est le DTW pour ce traitement.

Avantages du DDTW

La similitude peut être calculée en tenant compte de formes telles que la tendance à la hausse et à la baisse

Inconvénients du DDTW

La moyenne des pistes est-elle bonne?

code

from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor
import time
import numpy as np
import pylab as plt
import seaborn as sns
from fastdtw import fastdtw
from scipy.spatial.distance import euclidean


def get_test_curves(view=False):
    T = 150
    t = .4

    A = np.sin(np.array(range(T)) / 10)
    B = np.sin((np.array(range(T)) / 10 + t * np.pi))
    C = np.zeros((T))
    if view:
        plt.plot(A)
        plt.plot(B)
        plt.plot(C)
        plt.legend(['sin(θ)', 'sin(θ+150*pi)', 'constant'], fontsize=10, loc=2)
        plt.show()

    return {'name': 'sin(θ)', 'data': A}, {'name': 'sin(θ+150*pi)', 'data': B}, {'name': 'constant', 'data': C}


def mse(a, b):
    return ((a-b)**2).mean()


def get_DWT_results(T, S, skip=1, view=False):
    T_data, S_data = T['data'], S['data']
    T_name, S_name = T['name'], S['name']
    distance, path = fastdtw(T_data, S_data, dist=euclidean)
    print("DTW({}, {}):".format(T_name, S_name), distance)
    if view:
        plt.plot(T_data)
        plt.plot(S_data)
        k = -1
        for i, j in path:
            k += 1
            if k % skip == 0:
                plt.plot([i, j], [T_data[i], S_data[j]],
                         color='gray', linestyle='dotted')
        plt.legend([T_name, S_name], fontsize=10, loc=2)
        plt.title('DWT plot result')
        plt.show()

    return distance, path


def get_derivative(T):
    diff = np.diff(T)
    next_diff = np.append(np.delete(diff, 0), 0)
    avg = (next_diff + diff) / 2
    avg += diff
    avg /= 2
    return np.delete(avg, -1)


def get_DDWT_results(T, S, skip=1, view=False):
    T_data, S_data = T['data'], S['data']
    dT_data = get_derivative(T_data)
    dS_data = get_derivative(S_data)
    T_name, S_name = T['name'], S['name']
    distance, path = fastdtw(dT_data, dS_data, dist=euclidean)
    print("DDTW({}, {}):".format(T_name, S_name), distance)
    if view:
        plt.plot(T_data)
        plt.plot(S_data)
        k = -1
        for i, j in path:
            k += 1
            if k % skip == 0:
                plt.plot([i+1, j+1], [T_data[i+1], S_data[j+1]],
                         color='gray', linestyle='dotted')
        plt.legend([T_name, S_name], fontsize=10, loc=2)
        plt.title('DDWT plot result')
        plt.show()

    return distance, path


def get_test_curves_DDTWvsDWT(view=False):
    T = 150
    t = .4
    A = np.zeros((T))
    B = np.zeros((T))
    # A = np.sin(np.array(range(T)) / 10)
    # B = np.sin(np.array(range(T)) / 10+2)+50
    s_i = 50
    e_i = 60
    for i in range(s_i, e_i, 1):
        A[i] = np.sin(np.pi*(i-s_i)/(e_i-s_i))
    #     B[i] = -2.2
    if view:
        plt.plot(A)
        plt.plot(B)
        plt.legend(['sin(θ)', 'sin(θ+150*pi)'], fontsize=10, loc=2)
        plt.show()

    return {'name': 'down', 'data': A}, {'name': 'up', 'data': B}


def main():
    print("=== main ===")
    # A, B, C = get_test_curves()
    A, B = get_test_curves_DDTWvsDWT()

    # A["data"] = np.array([2, 0, 1, 1, 2, 4, 2, 1, 2, 0])
    # B["data"] = np.array([1, 1, 2, 4, 2, 1, 2, 0])

    print("{}-{}:".format(A['name'], B['name']), mse(A['data'], B['data']))
    # print("{}-{}:".format(A['name'], C['name']), mse(A['data'], C['data']))
    #Calculer DTW
    get_DWT_results(A, B, skip=5)
    get_DDWT_results(A, B, skip=5)


if __name__ == "__main__":
    main()

Article appliqué

Regroupement des fluctuations des cours boursiers de 2 940 entreprises japonaises dans DTW (Dynamic Time Warping) Dynamic Time Warping by Kimiaki Shirahama Comparez la différence de résultats en utilisant DTW et DDTW. Les données utilisées sont «Thomson Reuters Data Stream». Nous concluons par une évaluation qualitative (visuellement) que DTW + DDTW est la mesure de similarité optimale pour classer et analyser les données sur les cours des actions.

Référence 3 est très utile avec le code et les expériences. S'applique aux données de température.

De plus, il semble y avoir quelque chose qui s'appelle k-shape. Je n'ai pas lu ceci, mais j'utilise la distance comme corrélation mutuelle et le cluster en utilisant k-means.

Impression

J'ai en fait essayé de regrouper les données de séries chronologiques à l'aide de DTW / DDTW, mais cela n'a pas fonctionné. La cause est que je ne comprends pas grand-chose. Si vous avez des suggestions, n'hésitez pas à nous contacter.

[PYTHON] Comment comparer des données de séries chronologiques - Dérivée DTW, DTW-

emballer

** Comment trouver DTW **

** Avantages de DTW **

** Inconvénients de DTW **

** Avantages du DDTW **

** Inconvénients du DDTW **

** Article appliqué **

** Impression **