dtaidistance
v2.3.10
Biblioteca para distâncias de séries temporais (por exemplo, Dynamic Time Warping) usada no Grupo de Pesquisa DTAI. A biblioteca oferece uma implementação Python pura e uma implementação rápida em C. A implementação C tem apenas Cython como dependência. É compatível com Numpy e Pandas e implementado de forma que operações desnecessárias de cópia de dados sejam evitadas.
Documentação: http://dtaidistance.readthedocs.io
Exemplo:
from dtaidistance import dtw
import numpy as np
s1 = np.array([0.0, 0, 1, 2, 1, 0, 1, 0, 0])
s2 = np.array([0.0, 1, 2, 0, 0, 0, 0, 0, 0])
d = dtw.distance_fast(s1, s2)
Citando este trabalho:
Wannes Meert, Kilian Hendrickx, Toon Van Craenendonck, Pieter Robberechts, Hendrik Blockeel e Jesse Davis.
DTAIDistance (versão v2). Zenodo.
http://doi.org/10.5281/zenodo.5901139
Novidade na v2 :
ssize_t em vez de int permite estruturas de dados maiores em máquinas de 64 bits e é mais compatível com Numpy.max_dist revelou-se semelhante ao trabalho de Silva e Batista no PrunedDTW [7]. A caixa de ferramentas agora implementa uma versão igual ao PrunedDTW, pois remove mais distâncias parciais. Além disso, um argumento use_pruning é adicionado para definir automaticamente max_dist para a distância euclidiana, conforme sugerido por Silva e Batista, para acelerar o cálculo (um novo método ub_euclidean está disponível).dtaidistance.dtw_ndim . $ pip install dtaidistance
ou
$ conda install -c conda-forge dtaidistance
A instalação do pip requer Numpy como uma dependência para compilar código C compatível com Numpy (usando Cython). No entanto, esta dependência é opcional e pode ser removida.
O código-fonte está disponível em github.com/wannesm/dtaidistance.
Se você encontrar algum problema durante a compilação (por exemplo, a implementação baseada em C ou OpenMP não está disponível), consulte a documentação para mais opções.
from dtaidistance import dtw
from dtaidistance import dtw_visualisation as dtwvis
import numpy as np
s1 = np.array([0., 0, 1, 2, 1, 0, 1, 0, 0, 2, 1, 0, 0])
s2 = np.array([0., 1, 2, 3, 1, 0, 0, 0, 2, 1, 0, 0, 0])
path = dtw.warping_path(s1, s2)
dtwvis.plot_warping(s1, s2, path, filename="warp.png")
Apenas a medida de distância baseada em duas sequências de números:
from dtaidistance import dtw
s1 = [0, 0, 1, 2, 1, 0, 1, 0, 0]
s2 = [0, 1, 2, 0, 0, 0, 0, 0, 0]
distance = dtw.distance(s1, s2)
print(distance)
A versão mais rápida (30-300 vezes) usa c diretamente, mas requer um array como entrada (com o tipo double) e (opcionalmente) também remove os cálculos definindo max_dist para o limite superior euclidiano:
from dtaidistance import dtw
import array
s1 = array.array('d',[0, 0, 1, 2, 1, 0, 1, 0, 0])
s2 = array.array('d',[0, 1, 2, 0, 0, 0, 0, 0, 0])
d = dtw.distance_fast(s1, s2, use_pruning=True)
Ou você pode usar um array numpy (com dtype double ou float):
from dtaidistance import dtw
import numpy as np
s1 = np.array([0, 0, 1, 2, 1, 0, 1, 0, 0], dtype=np.double)
s2 = np.array([0.0, 1, 2, 0, 0, 0, 0, 0, 0])
d = dtw.distance_fast(s1, s2, use_pruning=True)
Verifique o __doc__ para obter informações sobre os argumentos disponíveis:
print(dtw.distance.__doc__)
Estão previstas uma série de opções para interromper antecipadamente alguns caminhos que o algoritmo de programação dinâmica está explorando ou ajustar o cálculo da medida de distância:
window : permite apenas deslocamentos até esse valor longe das duas diagonais.max_dist : Pare se a medida de distância retornada for maior que este valor.max_step : Não permita etapas maiores que este valor.max_length_diff : Retorna infinito se a diferença no comprimento de duas séries for maior.penalty : Penalidade a ser adicionada se compressão ou expansão for aplicada (no topo da distância).psi : Relaxamento Psi para ignorar início e/ou fim de sequências (para sequências cíclicas) [2].use_pruning : remove cálculos com base no limite superior euclidiano. Se, além da distância, você também quiser que a matriz completa veja todos os caminhos de distorção possíveis:
from dtaidistance import dtw
s1 = [0, 0, 1, 2, 1, 0, 1, 0, 0]
s2 = [0, 1, 2, 0, 0, 0, 0, 0, 0]
distance, paths = dtw.warping_paths(s1, s2)
print(distance)
print(paths)
A matriz com todos os caminhos de warping pode ser visualizada da seguinte forma:
from dtaidistance import dtw
from dtaidistance import dtw_visualisation as dtwvis
import random
import numpy as np
x = np.arange(0, 20, .5)
s1 = np.sin(x)
s2 = np.sin(x - 1)
random.seed(1)
for idx in range(len(s2)):
if random.random() < 0.05:
s2[idx] += (random.random() - 0.5) / 2
d, paths = dtw.warping_paths(s1, s2, window=25, psi=2)
best_path = dtw.best_path(paths)
dtwvis.plot_warpingpaths(s1, s2, paths, best_path)
Observe o parâmetro psi que relaxa a correspondência no início e no final. Neste exemplo, isso resulta em uma correspondência perfeita, mesmo que as ondas senoidais estejam ligeiramente deslocadas.
Para calcular as medidas de distância DTW entre todas as sequências em uma lista de sequências, use o método dtw.distance_matrix . Você pode definir variáveis para usar mais ou menos código c ( use_c e use_nogil ) e execução paralela ou serial ( parallel ).
O método distance_matrix espera uma lista de listas/arrays:
from dtaidistance import dtw
import numpy as np
series = [
np.array([0, 0, 1, 2, 1, 0, 1, 0, 0], dtype=np.double),
np.array([0.0, 1, 2, 0, 0, 0, 0, 0, 0, 0, 0]),
np.array([0.0, 0, 1, 2, 1, 0, 0, 0])]
ds = dtw.distance_matrix_fast(series)
ou uma matriz (caso todas as séries tenham o mesmo comprimento):
from dtaidistance import dtw
import numpy as np
series = np.matrix([
[0.0, 0, 1, 2, 1, 0, 1, 0, 0],
[0.0, 1, 2, 0, 0, 0, 0, 0, 0],
[0.0, 0, 1, 2, 1, 0, 0, 0, 0]])
ds = dtw.distance_matrix_fast(series)
Você pode instruir o cálculo para preencher apenas parte da matriz de medidas de distância. Por exemplo, para distribuir os cálculos em vários nós ou para comparar apenas as séries de origem com as séries de destino.
from dtaidistance import dtw
import numpy as np
series = np.matrix([
[0., 0, 1, 2, 1, 0, 1, 0, 0],
[0., 1, 2, 0, 0, 0, 0, 0, 0],
[1., 2, 0, 0, 0, 0, 0, 1, 1],
[0., 0, 1, 2, 1, 0, 1, 0, 0],
[0., 1, 2, 0, 0, 0, 0, 0, 0],
[1., 2, 0, 0, 0, 0, 0, 1, 1]])
ds = dtw.distance_matrix_fast(series, block=((1, 4), (3, 5)))
A saída neste caso será:
# 0 1 2 3 4 5
[[ inf inf inf inf inf inf] # 0
[ inf inf inf 1.4142 0.0000 inf] # 1
[ inf inf inf 2.2360 1.7320 inf] # 2
[ inf inf inf inf 1.4142 inf] # 3
[ inf inf inf inf inf inf] # 4
[ inf inf inf inf inf inf]] # 5
Uma matriz de distância pode ser usada para agrupamento de séries temporais. Você pode usar métodos existentes, como scipy.cluster.hierarchy.linkage ou um dos dois métodos de cluster incluídos (o último é um wrapper para o método de ligação SciPy).
from dtaidistance import clustering
# Custom Hierarchical clustering
model1 = clustering.Hierarchical(dtw.distance_matrix_fast, {})
cluster_idx = model1.fit(series)
# Augment Hierarchical object to keep track of the full tree
model2 = clustering.HierarchicalTree(model1)
cluster_idx = model2.fit(series)
# SciPy linkage clustering
model3 = clustering.LinkageTree(dtw.distance_matrix_fast, {})
cluster_idx = model3.fit(series)
Para modelos que acompanham a árvore de cluster completa ( HierarchicalTree ou LinkageTree ), a árvore pode ser visualizada:
model.plot("myplot.png")
Opcional:
Desenvolvimento:
DTAI distance code.
Copyright 2016-2022 KU Leuven, DTAI Research Group
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
