AIによる異常検知 勤務データ編

目次

1.はじめに
2.オートエンコーダーを用いた異常検知
3.前準備
4.学習
5.検証
6.おわりに

1. はじめに

AIを活用した異常検知は、いろいろな場面で活用されています。

例えば、製造工場やエネルギー設備などの産業設備において、異常検知は非常に重要です。センサーデータや機械の動作データを監視し、通常の動作パターンからの逸脱を検知することで、機械の故障や異常を予測し、メンテナンスを行うことができます。

また、医療分野では、患者の健康状態をモニタリングするために異常検知が活用されます。患者の生体データ（心拍数、血圧、体温など）を収集し、個々の患者に合わせた基準と比較して異常を検出します。また、画像診断においても、異常な組織や器官の特徴を検出するためにAI技術が利用されます。

この記事では、どこの企業にでもある勤務データを活用して、AI（オートエンコーダー）による実践的な異常検知を試行してみようと思います。

2. オートエンコーダーを用いた異常検知

オートエンコーダーは、入力データを低次元の表現にエンコードし、それを再び元の形式にデコードするニューラルネットワークの一種です。一般的な構造は、エンコーダーとデコーダーで構成されており、中間層には「ボトルネック」と呼ばれる低次元表現が存在します。

異常検知では、まず正常なデータ（例: 勤務データの正常なパターン）でオートエンコーダーモデルを学習させます。この際、正常なデータを入力として与え、エンコーダーで低次元表現に変換し、それをデコーダーで元の次元に戻すように学習します。このとき、モデルは正常なデータのパターンを学習し、そのパターンを再構築できるようになります。

学習が完了した後、未知のデータ（新たな勤務データ）をモデルに入力します。モデルはそのデータをエンコードし、デコードして再構築を行います。正常なデータに似たものは正常に再構築できますが、異常なデータは正常なパターンから逸脱しているため、再構築がうまく行かないことがあります。

ここで、再構築誤差（元データと再構築データの差）が重要です。正常なデータは再構築誤差が小さくなる傾向がありますが、異常なデータは再構築誤差が大きくなります。この再構築誤差を利用して、異常かどうかを判定します。閾値を設定し、再構築誤差が閾値を超える場合には、そのデータは異常と判定されます。再構築誤差は、ここでは平均二乗誤差により、算出しています。

オートエンコーダーは正常なデータのパターンを学習し、それに基づいて未知のデータの再構築を行う能力を持っています。異常なデータは正常なパターンから逸脱しているため、その再構築誤差が大きくなり、それを通じて異常を検出する仕組みとなります。

3. 前準備

まずは、勤務データですが、実際のデータを使用するわけにはいかないので、ダミーデータとして正常データを作成します。

勤務データの基本的なパターンは、以下の通りです。各１日毎に４件のデータです。
A.PC立ち上げ時刻
B.勤務開始時刻
C.勤務終了時刻
D.PCシャットダウン時刻

４日分の勤務データで、ワンセットとします。4*4=16となります。
コンピュータは、ご存じの通り、０と１の２進法の世界です。
２の階乗のデータ数と相性が良いので、2^4=16個のデータでワンセットとします。

import pandas as pd
import random
from datetime import datetime, timedelta

# データ数
data_count = 100

# 時刻の範囲と割合
eight_o_clock_percentage = 0.8  # 8時台の割合
seven_o_clock_percentage = 0.1  # 7時台の割合
nine_to_twelve_percentage = 0.1  # 9時台から12時台の割合
end_time_range = (17, 22)
start_end_overlap = 30  # 分
a_b_overlap = 30  # 分
c_d_overlap = 30  # 分
start_time_distribution = [0.1, 0.9]
end_time_distribution = [0.4, 0.6]

# データ生成
data = []
for _ in range(data_count):
    day_data = []  # 1行分の4日分のデータを格納するリスト

    for _ in range(4):
        # A.PC立ち上げ時刻
        pc_start_hour = None
        random_percentage = random.random()
        if random_percentage < eight_o_clock_percentage:
            pc_start_hour = random.choice(range(8, 9))  # 8時台
        elif random_percentage < eight_o_clock_percentage + seven_o_clock_percentage:
            pc_start_hour = random.choice(range(7, 8))  # 7時台
        else:
            pc_start_hour = random.choice(range(9, 13))  # 9時台から12時台

        pc_start_time = datetime.strptime(f"{pc_start_hour:02d}:00", "%H:%M")

        # B.勤務開始時刻
        work_start_time = pc_start_time + timedelta(minutes=a_b_overlap)
        work_start_time = work_start_time.replace(minute=random.choice(range(0, 30, 5)))

        # C.勤務終了時刻
        work_end_time = datetime.strptime(f"{random.choice(range(end_time_range[0], end_time_range[1])):02d}:00", "%H:%M")
        work_end_time = work_end_time.replace(minute=random.choice(range(0, 30, 5)))

        # D.PCシャットダウン時刻
        pc_shutdown_time = work_end_time + timedelta(minutes=c_d_overlap)

        day_data.extend([pc_start_time.strftime("%H:%M"), work_start_time.strftime("%H:%M"), work_end_time.strftime("%H:%M"), pc_shutdown_time.strftime("%H:%M")])

    data.append(day_data)

# CSVファイルに保存
columns = ['PC立ち上げ時刻', '勤務開始時刻', '勤務終了時刻', 'PCシャットダウン時刻'] * 4  # 4日分なので4回繰り返す
df = pd.DataFrame(data, columns=columns)
df.to_csv('work_data.csv', index=False)

# データフレーム表示
df.head()

ダミーデータは、以下のようになっています。
各行の先頭１日分のデータは、以下の通りです。

実際は、４日分の勤務データがワンセットで以下のようになっています。全部で、100行分あります。

4. 学習

データが手に入りましたので、いよいよ、学習です。

簡単に、コードの概要解説をします。

Sequentialを使います。Kerasという深層学習フレームワーク内のモデル定義のためのクラスです。Kerasは、ニューラルネットワークモデルの構築やトレーニングを簡単に行うためのツールセットです。

Sequentialを使うことで、モデルを一つのスタックとして定義できます。つまり、層を順番に積み重ねていくことで、ネットワークを作成することができます。これは、シンプルな層の積み重ねの場合や、順序が重要なネットワークの場合に便利です。

model.add(...)の部分には、畳み込み層、全結合層、プーリング層など、様々な種類のニューラルネットワーク層を追加することができます。層を追加することで、モデルの構造が徐々に形成されていきます。後ほど、詳述します。

model.compile(...)は、モデルの学習のために必要な損失関数や最適化アルゴリズム、評価指標などを指定します。こちらについても、後ほど、詳述します。

import keras.models
from keras.layers import Dense, BatchNormalization, Activation
from keras.models import Sequential

ModelFile = "ae_model.h5"

# 入力データの項目数
INPUT_SIZE = 16

# ダミーデータの読み込み
train_data = pd.read_csv('work_data.csv')

# Convert time strings to numeric features
def convert_time_to_minutes(time_str):
    hours, minutes = map(int, time_str.split(':'))
    return hours * 60 + minutes

for column in train_data.columns:
    train_data[column] = train_data[column].apply(convert_time_to_minutes)

# モデルのインスタンス化
model = Sequential()

# ポイント１．オートエンコーダーの構築ブロック
# 入力層
model.add(Dense(INPUT_SIZE,input_shape=(INPUT_SIZE,)))
model.add(BatchNormalization())
model.add(Activation('relu'))

# 中間層
model.add(Dense(int(INPUT_SIZE/2)))
model.add(BatchNormalization())
model.add(Activation('relu'))

# 出力層
model.add(Dense(INPUT_SIZE))
model.add(BatchNormalization())
model.add(Activation('relu'))

model.add(Dense(INPUT_SIZE))
model.compile(optimizer='adam',
                       loss='mean_squared_error')
model.summary()

# ポイント２．学習ブロック
history = model.fit(x=train_data, y=train_data,
                             epochs=1000,
                             batch_size=4,
                             validation_split=0.1)

# モデルを保存
model.save(ModelFile)

コードにおけるポイントは、２つです。

A. ポイント１：オートエンコーダーの構築ブロック

１つのポイントである、オートエンコーダーの構築ブロックについて、説明します。

まず、前半部分です。入力層、中間層、出力層ともに、３つの部分、Dense（全結合層）、BatchNormalization（バッチ正規化）、Activation（活性化関数）から、成り立っています。

Dense（全結合層）
Denseは、ニューラルネットワークの基本的な層の一つです。入力ユニットと出力ユニットが完全に結合しており、各入力ユニットの情報が全ての出力ユニットに伝播されます。これは、重み行列とバイアスベクトルによって特徴量の変換を行います。オートエンコーダーでは、エンコーダーやデコーダーの中間層にDense層が使われ、入力データの次元を変換する役割を果たします。

BatchNormalization（バッチ正規化）
バッチ正規化は、ニューラルネットワークの学習を安定化させるために使われる手法です。ニューラルネットワークの中間層の出力を正規化し、平均を0、分散を1に近づけることで、学習が効率的に行えるようになります。バッチ正規化は、層の間でのデータの分布の偏りを軽減し、勾配消失問題を緩和するのに役立ちます。オートエンコーダーでも、中間層でバッチ正規化を適用することで、学習の収束を改善できます。

Activation（活性化関数）
活性化関数は、ニューラルネットワークの各層の出力を非線形に変換するために使用されます。線形変換だけでは、ネットワーク全体が一つの線形変換としてまとめられてしまい、多層構造の恩恵を受けることができません。代表的な活性化関数には、ReLU（Rectified Linear Unit）、Sigmoid、Tanhなどがあります。上記コードでは、収束が早いReLUを使っています。これらの関数は、入力に対して非線形な変換を行い、ネットワークがより複雑な関数を学習できるようにします。オートエンコーダーでは、エンコーダーやデコーダーの各層で活性化関数を適用して、特徴の抽出や再構築を行います。

次に、オートエンコーダーの構築ブロックの後半部分について、説明します。

model.compile(optimizer='adam', loss='mean_squared_error')は、Kerasモデルのトレーニングに関する設定を行うメソッドです。具体的には、モデルの最適化アルゴリズム（optimizer）と損失関数（loss）を指定します。

optimizer（最適化アルゴリズム）
最適化アルゴリズムは、モデルのパラメータ（重みとバイアスなど）を調整して、モデルが訓練データに適応するように更新するアルゴリズムです。ここで使用している"adam"は、一般的な最適化アルゴリズムの一つで、特に深層学習モデルでよく使われます。"adam"は勾配降下法のバリエーションであり、適応的な学習率の調整を行うことで、学習の収束を早めることができます。他にも"sgd"（確率的勾配降下法）や"rmsprop"（RMSProp最適化）などもよく使用されます。

loss（損失関数）
損失関数は、モデルの出力と正解データの間の誤差を評価する関数です。訓練中にこの誤差を最小化することで、モデルは正しい予測を行えるようになります。"mean_squared_error"（平均二乗誤差）は、予測と正解の間の差の二乗を取り、それらの平均を計算する損失関数です。オートエンコーダーなど、入力データと出力データが似ている場合に適した損失関数です。他にも、分類タスクでは"categorical_crossentropy"（カテゴリカルクロスエントロピー）、"binary_crossentropy"（バイナリクロスエントロピー）などがあります。

model.compile(optimizer='adam', loss='mean_squared_error')は、モデルが"adam"最適化アルゴリズムを使用して訓練され、訓練中に平均二乗誤差を最小化しようとすることを指定しています。この設定は、オートエンコーダーなどの出力と入力が似ている場合に一般的に適しています。

model.summary() は、Kerasモデルの概要を表示するためのメソッドです。モデルの構造や各層のパラメータ数、出力形状などの詳細な情報を簡潔にまとめて表示します。これは、モデルのアーキテクチャを理解し、ネットワークの構造を確認するために非常に役立つツールです。

B. ポイント２：学習ブロック

もう１つのポイントである、学習ブロックについて、説明します。

history = model.fit(x=train_data, y=train_data, epochs=1000, batch_size=4, validation_split=0.1)

historyオブジェクトには、エポックごとの訓練と検証の損失および評価指標が格納され、トレーニングプロセスの進行状況を視覚化するのに役立ちます。

fitは、Kerasや他の深層学習フレームワークで使用されるメソッドの一つで、モデルの訓練プロセスを開始するために使用されます。モデルを訓練する際には、fitメソッドを呼び出して訓練データを与え、モデルの重みやパラメータを最適化して目標とする出力を得るように学習させます。

パラメータは以下の通りです。

train_data: 訓練データの入力と出力のペアを表すNumPy配列やテンソルです。モデルは、この入力データに対して適切な出力を生成するように学習します。

epochs: エポック数を指定します。エポックは、訓練データを一巡する単位です。モデルは各エポックで訓練データを使用して学習を行い、指定されたエポック数だけ訓練を繰り返します。

batch_size: バッチサイズを指定します。バッチサイズは、一度にモデルに提供される訓練データの数です。訓練データをバッチに分割して、各バッチごとにモデルのパラメータを更新します。小さなバッチサイズはメモリの使用効率を向上させますが、計算の効率性に影響を与えることがあります。

validation_split: 検証データを訓練データから分割する割合を指定します。この値に基づいて、訓練中にモデルの性能を検証します。例えば、validation_split=0.1の場合、訓練データの10%が検証データとして使用されます。

5. 検証

学習したモデルを使って、検証してみます。
５件のダミーデータを使って、検証を行います。見て頂ければ分かる通り、PC立ち上げ時刻が５時とかになっていて、いかにも怪しいデータです(笑)

ちなみに、正常・異常を判断する閾値は、事前に正常データと異常データで試して、境界になりそうな値を使用しています。

import numpy as np
from keras.models import load_model
from sklearn.metrics import mean_squared_error

# 学習済みモデルの読み込み
ModelFile = "ae_model.h5"
model = load_model(ModelFile)

# 判別対象のデータ
data_strings = [
    '08:00,08:20,17:00,17:30,08:00,08:25,18:00,18:30,08:00,08:15,18:25,18:55,11:00,11:15,19:10,19:40',
    '05:00,05:20,23:00,23:30,07:00,07:25,20:00,20:30,08:00,08:15,18:25,18:55,11:00,11:15,19:10,19:40',
    '07:00,07:20,17:00,17:30,07:00,07:25,20:00,20:30,08:00,08:15,18:25,18:55,11:00,11:15,19:10,19:40',
    '06:00,06:20,22:00,22:30,04:00,04:25,24:00,24:30,08:00,08:15,18:25,18:55,11:00,11:15,19:10,19:40',
    '07:00,07:20,17:00,17:30,07:00,07:25,20:00,20:30,08:00,08:15,18:25,18:55,08:00,08:20,19:10,19:40'
]

# データの前処理
def preprocess_data(data_strings):
    data = []
    for data_str in data_strings:
        times = data_str.split(',')
        data_row = []
        for time in times:
            hours, minutes = map(int, time.split(':'))
            data_row.append(hours * 60 + minutes)
        data.append(data_row)
    return np.array(data)

# データの前処理
test_data = preprocess_data(data_strings)

# 再構築誤差を計算
reconstructed_data = model.predict(test_data)
errors = np.mean(np.square(test_data - reconstructed_data), axis=1)

# 閾値を用いて異常か正常かを判別
threshold = 100000
predictions = ['異常' if error > threshold else '正常' for error in errors]

# 平均二乗誤差を計算
detect_score = mean_squared_error(test_data, reconstructed_data)

# 結果とスコアを表示
for i, (prediction, error) in enumerate(zip(predictions, errors)):
    print(f"データ{i+1}: {prediction}, Mean Squared Error: {error:.2f}")

結果は、以下の通りです。想定通り、正常、異常を判別できています！

6. おわりに

如何だったでしょうか？
AI（オートエンコーダー）を使って、異常検知が出来ました！

実際のデータで、こんなにキレイに切り分けができるかどうかは、やってみないと分からない面もあります。しかしながら、オートエンコーダーを使った異常検知は、安定的な結果を得やすいAIと言われています。

今回は、勤務データで異常検知を実施しましたが、例えば、プロジェクト管理などもできるかもしれません。正常なプロジェクトのデータを学習させて、その後、検証したいプロジェクトのデータを入れることで、正常なプロジェクトとの乖離状況を、客観的に数値で把握できると良いですね。

オートエンコーダーはニューラルネットワークの一種です。2010年代以降、ディープラーニング技術の急速な進化と広まりが影響し、注目を浴びている比較的新しい技術です。ぜひとも、実務において、実践的に活用できるか、トライしても良いかと思います。

参考

「オートエンコーダー」と似ているAI用語に「エンコーダー・デコーダー・モデル」があります。両方とも機械学習やディープラーニングのコンセプトに関連する用語ですが、違いがあります。

エンコーダー・デコーダー・モデルは、Chat-GPTでも使われている技術です。参考までに、比較して、説明します。

オートエンコーダー (Autoencoder)
オートエンコーダーは、データの特徴を抽出するために使用されます。一般的には、入力データを低次元の表現にエンコードし、それを再び元の形式にデコードすることを目指します。オートエンコーダーは、主にデータの圧縮、ノイズ除去、特徴抽出などのタスクに使用されます。

エンコーダー・デコーダー・モデル (Encoder-Decoder Model)
エンコーダー・デコーダー・モデルは、主にシーケンスデータや変換タスクに関連しています。このアーキテクチャは、エンコーダーとデコーダーがペアとなって、一つのタイプのデータを別のタイプに変換するために使用されます。典型的な例としては、機械翻訳（文章の翻訳）、音声合成、画像キャプショニング（画像に対する文章の生成）などがあります。