Prophetを使ってみた（異常検知） #Python3

今回は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にコードを記述して定期的に実行＋異常を検知した場合のみメール送信するなどのアクションを組み合わせる事もできますので、便利だなーと感じました。