LoginSignup
3
1

More than 1 year has passed since last update.

Prophetを使ってみた(異常検知)

Posted at

今回は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. 実行結果

こんな感じで異常検知されていることが確認できます。
image.png

マウスカーソルを合わせる事で値を表示させることができるので、いい感じですね。
image.png

6. その他

今回はjupyter notebook上で実行しましたが、pythonにコードを記述して定期的に実行+異常を検知した場合のみメール送信するなどのアクションを組み合わせる事もできますので、便利だなーと感じました。

3
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
3
1