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
La similitude peut être obtenue même si les longueurs et les cycles des séries chronologiques sont différents.
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.
La similitude peut être calculée en tenant compte de formes telles que la tendance à la hausse et à la baisse
La moyenne des pistes est-elle bonne?
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()
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.
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.
Recommended Posts