【異常検知】lof法 - えんじにあのじゆうちょう

はじめに

前回の最近傍法を使った方法では、「稀によくある」ようなものを異常としたくない場合をうまくハンドリングできないということを書きました。
https://t.marufeuille.dev/entory/nearest-neighbort.marufeuille.dev

今回はlof法を用いた異常検知を解説します。

解説

概要

lof法は距離をベースとするという点では最近傍法と同じですが、「最も近い点に対しての距離」と「見つけた最も近い点から最も近い点の距離」の比をもって異常値の判断を行います。

つまり、時系列で1, 2, 1.5, 10, 1, 1, 1, 10, 1, 1...というデータ系列の時、閾値を3とかにすると、途中に出てくる10は異常値という扱いですが、これの場合、1つ目の10に最も近い点は10となり、その距離は4、その10から最も近い点は元々の10となり、距離は4で比を取ると1となります。
これに対して閾値を求めることで、密度をベースとした異常検知が行なえます。

実装

データの生成については例のごとく、2回前の記事を参照してください。
https://t.marufeuille.dev/entory/hotelings-theoryt.marufeuille.dev

念の為、データはこんな感じです。
f:id:marufeuillex:20191214205628p:plain

# 距離を計算する関数
def get_neighbor(x, series):
    min_val = np.inf
    idx = 0
    for i in range(series.shape[0]):
        if x[0] == p[i][0] and x[1] == p[i][1]:
            continue
        diff = np.linalg.norm(x - p[i])
        if diff < min_val:
            min_val = diff
            idx = i
    return (idx, min_val)

# 異常値検知
anomalies = []
for idx in range(NUM_SERIES):
    series = series_agg[idx]
    p = [np.array([i, series[i]]) for i in range(series.shape[0])]

    neighbors = [get_neighbor(p[i], series) for i in range(series.shape[0])]
    thresh = 3.0
    anomaly = []
    for i in range(series.shape[0]):
        min_val = np.inf

        for j in range(series.shape[0]):
            if i == j:
                continue
            diff = np.linalg.norm(p[i] - p[j])
            if diff < min_val:
                min_val = diff
        
        if min_val / neighbors[neighbors[i][0]][1] > thresh:
            anomaly.append(i)
    anomalies.append(anomaly)

# 可視化
fig, ax = plt.subplots(3, 2, figsize=(15, 10))
for i in range(len(series)):
    anomaly = np.array(anomalies[i])
    s = series[i]
    ax[i //2, i % 2].plot(np.arange(N), series[i])
    if len(anomaly) != 0:
        anomaly_data = np.array(s)[anomaly]
        ax[i //2, i % 2].scatter(anomaly, anomaly_data, color="red", marker="x")