使用動態時間規整來同步時間序列資料

介紹

在資料相關的職業生涯中遇到最痛苦的事情之一就是必須處理不同步的時間序列資料集。差異可能是由許多原因造成的——日光節約調整、不準確的SCADA訊號和損壞的資料等等。在相同的資料集中，在不同的點上發現幾個差異是很常見的，這需要分別識別和糾正每一個差異。而且當使用它時，可能會無意中抵消另一個同步部分。幸運的是，在新的“動態時間規整”技術的幫助下，我們能夠對所有的非同步資料集應用一種適用於所有解決方案。

動態時間規整

簡稱DTW是一種計算兩個資料序列之間的最佳匹配的技術。換句話說，如果你正在尋找在任何給定時間從一個數據集到另一個數據集的最短路徑。這種方法的美妙之處在於它允許你根據需要對資料集應用盡可能多的校正，以確保每個點都儘可能同步。，甚至可以將其應用於不同長度的資料集。 DTW 的應用是無窮無盡的，可以將它用於時間和非時間資料，例如財務指標、股票市場指數、計算音訊等。唯一的警告是確保資料沒有空值或缺失值，因為這可能會給 DTW 的工作帶來麻煩。

用於尋找對應點之間最短路徑的距離度量可以是 Scipy 的距離度量模組提供的任何度量。雖然在大多數情況下，歐幾里得距離可以解決問題，但是你可能希望與其他距離進行試驗以獲得良好的度量。

實現

為了實現我們自己的 DTW 版本，我們將使用 Python 中的 fastdtw 庫。這個包的新穎之處在於它簡化了扭曲函式的複雜性，從而將複雜性從 O（n²）降低到 O（n），這在執行時提供了明顯的差異。

請繼續啟動 Anaconda 或您選擇的任何 Python IDE 並安裝 fastdtw，如下所示：

pip install fastdtw

隨後，匯入所有必需的包：

import numpy as npimport pandas as pdimport streamlit as stimport plotly。express as pxfrom sklearn。metrics import r2_score

在運行同步之前匯入資料集並填寫任何空值：

df = pd。read_csv（‘dataset。csv’）df［‘Power’］ = pd。to_numeric（df［‘Power’］，errors=‘coerce’）df［‘Voltage’］ = pd。to_numeric（df［‘Voltage’］，errors=‘coerce’）x = np。array（df［‘Power’］。fillna（0））y = np。array（df［‘Voltage’］。fillna（0））

然後繼續執行同步：

distance， path = fastdtw（x， y， dist=euclidean）

同步路徑的結果將類似於以下內容：

path = ［（0， 0），（0， 1），（0， 2），（1， 3），（2， 4），。。。］

參考資料集中的每個點都將與目標資料集中的一個或多個點進行匹配，即參考資料的第 0 行可以與目標資料的點 0、1 或 2 匹配。

現在有了扭曲的路徑，可以繼續建立具有同步結果的資料框，如下所示：

result = ［］for i in range（0，len（path））：result。append（［df［‘DateTime’］。iloc［path［i］［0］］，df［‘Power’］。iloc［path［i］［0］］，df［‘Voltage’］。iloc［path［i］［1］］］）df_sync = pd。DataFrame（data=result，columns=［‘DateTime’，‘Power’，‘Voltage’］）。dropna（）df_sync = df_sync。drop_duplicates（subset=［‘DateTime’］）df_sync = df_sync。sort_values（by=‘DateTime’）df_sync = df_sync。reset_index（drop=True）df_sync。to_csv（‘C：/Users/。。。/synchronized_dataset。csv’，index=False）

最後，您可以使用 Sklearn 的 r2_score 模組計算相關性分數，以比較同步前後的相關性或同步水平：

correlation = r2_score（df［‘Power’］，df［‘Voltage’］）

資料視覺化

為了繪製和視覺化您的同步資料，我們將使用 Plotly 和 Streamlit——我最喜歡的兩個用於視覺化資料並將其呈現為應用程式的庫。

可以使用下面的函式來建立時間序列圖表。請確保時間戳採用正確的 dd-mm-yyyy hh：mm 格式，或者修改函式以適應你的資料。

def chart（df）：df_columns = list（df）df［‘DateTime’］ = pd。to_datetime（df［‘DateTime’］，format=‘%d-%m-%y %H：%M’）df［‘DateTime’］ = df［‘DateTime’］。dt。strftime（‘ %H：%M on %B %-d， %Y’）df = df。sort_values（by=‘DateTime’）fig = px。line（df， x=“DateTime”， y=df_columns，labels={“DateTime”： “DateTime”，“value”： “Value”，“variable”： “Variables”}，hover_data={“DateTime”： “|%d-%m-%Y %H：%M”}）fig。update_layout（font_family=“IBM Plex Sans”，font_color=“black”）fig。update_xaxes（rangeselector=dict（buttons=list（［dict（count=12， label=“12h”， step=“hour”， stepmode=“backward”），dict（count=1， label=“1d”， step=“day”， stepmode=“backward”），dict（count=7， label=“7d”， step=“day”， stepmode=“backward”），dict（step=“all”）］）））st。write（fig）

要視覺化和呈現圖表，請透過在 Anaconda 提示符下鍵入以下命令來執行您的指令碼：

cd C：/Users/。。。/local_directorystreamlit run synchronization。py

可以在同步之前和之後對資料進行視覺化：