今回はProphetを使った異常検知をやってみたいと思います。
予測検知の作業を実施した後をベースとしていますのでまだの方は、下記記事を先に見てもらえればと思います。
また、下記サイトを参考にさせていただきました。
https://www.kaggle.com/code/vinayjaju/anomaly-detection-using-facebook-s-prophet
1. ライブラリインストール
まずは、追加で下記ライブラリをインストールします。
conda install -c conda-forge altair
2. ライブラリインポート
では、必要となるライブラリをインポートしましょう。
import pandas as pd
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import altair as alt
from fbprophet import Prophet
3. データ読込/データクリーニング
前回同様、為替レート(日本円からドル)データを読込/クリーニングをします。
df = pd.read_csv("./data/DEXJPUS.csv")
df["DEXJPUS"] = df["DEXJPUS"].replace(".",None)
df["DEXJPUS"] = df["DEXJPUS"].astype(np.float64)
df["DATE"] = pd.to_datetime(df["DATE"])
df["ds"] = df["DATE"]
df["y"] = df["DEXJPUS"]
4. 関数定義
モデルをトレーニングするために、基本的なハイパーパラメーターをinterval_widthとchangepoint_rangeで定義します。これらは、境界の幅を調整するために使用できます。
def fit_predict_model(dataframe, interval_width = 0.99, changepoint_range = 0.8):
m = Prophet(daily_seasonality = False, yearly_seasonality = False, weekly_seasonality = False,
seasonality_mode = 'multiplicative',
interval_width = interval_width,
changepoint_range = changepoint_range)
m = m.fit(dataframe)
forecast = m.predict(dataframe)
forecast['fact'] = dataframe['y'].reset_index(drop = True)
return forecast
pred = fit_predict_model(df)
次に、モデル境界の上部より高く、下部をすべて外れ値(異常)として設定します。さらに、ドットが境界からどれだけ離れているかに基づいて、外れ値の重要度を設定します。
def detect_anomalies(forecast):
forecasted = forecast[['ds','trend', 'yhat', 'yhat_lower', 'yhat_upper', 'fact']].copy()
forecasted['anomaly'] = 0
forecasted.loc[forecasted['fact'] > forecasted['yhat_upper'], 'anomaly'] = 1
forecasted.loc[forecasted['fact'] < forecasted['yhat_lower'], 'anomaly'] = -1
forecasted['importance'] = 0
forecasted.loc[forecasted['anomaly'] ==1, 'importance'] = \
(forecasted['fact'] - forecasted['yhat_upper'])/forecast['fact']
forecasted.loc[forecasted['anomaly'] ==-1, 'importance'] = \
(forecasted['yhat_lower'] - forecasted['fact'])/forecast['fact']
return forecasted
pred = detect_anomalies(pred)
プロットを取得する準備が整いました。altairライブラリを利用してプロットしましょう。
def plot_anomalies(forecasted):
interval = alt.Chart(forecasted).mark_area(interpolate="basis", color = '#7FC97F').encode(
x=alt.X('ds:T', title ='date'),
y='yhat_upper',
y2='yhat_lower',
tooltip=['ds', 'fact', 'yhat_lower', 'yhat_upper']
).interactive().properties(
title='Anomaly Detection'
)
fact = alt.Chart(forecasted[forecasted.anomaly==0]).mark_circle(size=15, opacity=0.7, color = 'Black').encode(
x='ds:T',
y=alt.Y('fact', title='money order'),
tooltip=['ds', 'fact', 'yhat_lower', 'yhat_upper']
).interactive()
anomalies = alt.Chart(forecasted[forecasted.anomaly!=0]).mark_circle(size=30, color = 'Red').encode(
x='ds:T',
y=alt.Y('fact', title='money order'),
tooltip=['ds', 'fact', 'yhat_lower', 'yhat_upper'],
size = alt.Size( 'importance', legend=None)
).interactive()
return alt.layer(interval, fact, anomalies)\
.properties(width=870, height=450)\
.configure_title(fontSize=20)
plot_anomalies(pred)
5. 実行結果
マウスカーソルを合わせる事で値を表示させることができるので、いい感じですね。
6. その他
今回はjupyter notebook上で実行しましたが、pythonにコードを記述して定期的に実行+異常を検知した場合のみメール送信するなどのアクションを組み合わせる事もできますので、便利だなーと感じました。