100本ノック / 異常検知 / 異常検知100本ノック
製造業の異常検知を実務で使う｜評価指標・しきい値・アラート設計10本ノック
製造設備の異常を「行動できるアラート」に変える：評価指標とアラート設計10本ノック
異常検知モデルは、異常スコアを出せただけでは現場で価値を生みません。本記事では、架空の加工工場における設備監視を題材に、モデルの良し悪しを測る指標と、限られた保全工数をどのアラートへ割り当てるかを一つの意思決定問題として扱います。対象は No.081〜No.090 です。
[!NOTE] 本資料は、数理工房 (もしくは代表である和山個人) が過去に企業研修において使用した notebook を企業様の許可を得て再構成・編集のうえ公開しています。
掲載データはすべて架空のものであり、実在する企業・工場・数値とは一切関係ありません。
はじめに：この記事で扱う製造業の実務課題
加工設備から5分ごとに異常スコアが届く工場を想定します。真の故障兆候は少なく、点検できる件数にも上限があります。この状況で必要なのは「精度が高いモデル」だけではなく、見逃しによる停止損失、誤検知による点検工数、設備の重要度、対応期限を同時に考えた運用設計です。
本記事の到達点は、技術指標を報告することではありません。どのしきい値で、何件を、どの順に、誰が確認するかを説明できる状態を作ることです。
現場でよくある状況
- 異常は全観測の数％以下で、Accuracyだけを見ると常に「正常」と判定するモデルも高得点になる
- 誤検知が増えると、点検担当者がアラートを信用しなくなる
- 見逃しは突発停止、仕掛品廃棄、納期遅延につながるが、設備ごとに損失が異なる
- 同じ異常スコアでも、重要設備と補助設備では対応順が異なる
- モデル評価と現場のアラート管理表が分断されている
なぜこの問題は判断が難しいのか
Precisionを上げるためにしきい値を高くすると誤検知は減りますが、Recallが下がり見逃しが増えます。逆方向にも同じトレードオフがあります。F1、ROC-AUC、PR-AUCはいずれも有用ですが、金額、要員、対応期限を直接は表しません。そのため、統計指標、業務コスト、処理能力を段階的につなぐ必要があります。
今回扱うノックの全体像
| No. | テーマ | 意思決定への接続 |
|---|---|---|
| 081 | 評価指標 | 混同行列から判断の全体像を作る |
| 082 | Precision・Recall | 誤検知と見逃しのバランスを見る |
| 083 | F1スコア | 両者を一つの指標で比較する |
| 084 | ROC-AUC | 全しきい値での識別順位を確認する |
| 085 | PR-AUC | 希少な異常に対する実効性能を見る |
| 086 | 誤検知・見逃しコスト | 技術誤差を金額へ翻訳する |
| 087 | しきい値最適化 | 期待コストが小さい運用点を選ぶ |
| 088 | アラート件数 | 日別・ライン別の負荷を把握する |
| 089 | 優先度 | 重要度と緊急度で対応順を決める |
| 090 | 現場運用設計 | SLA、抑制、エスカレーションを定める |
Python 環境の準備
外部データは使わず、NumPy、pandas、Matplotlib、scikit-learnだけを利用します。乱数シードを固定し、同じ結果を再現できるようにします。グラフ内は環境依存の文字化けを避けるため英語表記とし、本文で日本語の読み取りを補います。
import sys
import numpy as np
import pandas as pd
import matplotlib
import matplotlib.pyplot as plt
from sklearn.metrics import (
confusion_matrix, precision_score, recall_score, f1_score,
roc_curve, roc_auc_score, precision_recall_curve, average_precision_score,
)
SEED = 202509
rng = np.random.default_rng(SEED)
pd.set_option("display.max_columns", 20)
pd.set_option("display.float_format", lambda x: f"{x:,.3f}")
print("Python :", sys.version.split()[0])
print("NumPy :", np.__version__)
print("pandas :", pd.__version__)
print("Matplotlib :", matplotlib.__version__)Python : 3.13.1
NumPy : 2.5.1
pandas : 3.0.3
Matplotlib : 3.11.0架空データの作成
30日間、3ライン・6設備から得た1,200件の判定機会を作ります。actual_anomaly は後日の点検・故障記録で確定した正解、anomaly_score はモデルが出した0〜1の異常度です。異常率を約6％に抑え、現実に近い不均衡を持たせます。さらに、設備重要度、停止影響額、検知からの経過時間を付与します。
n = 1200
timestamps = pd.date_range("2025-01-01", periods=n, freq="36min")
lines = rng.choice(["Line-A", "Line-B", "Line-C"], n, p=[0.40, 0.35, 0.25])
machines = np.array([f"{line}-M{rng.integers(1, 3)}" for line in lines])
actual = rng.binomial(1, 0.06, n)
# 正常と異常の分布を一部重ね、完全には判別できない状況を作る
score = np.where(actual == 1, rng.beta(5.0, 2.2, n), rng.beta(1.5, 6.0, n))
score = np.clip(score + np.where(lines == "Line-C", 0.025, 0), 0, 1)
criticality = np.where(lines == "Line-A", 3, np.where(lines == "Line-B", 2, 1))
impact_k_yen = criticality * rng.integers(350, 701, n) # 停止影響（千円）
elapsed_min = rng.integers(5, 181, n)
df = pd.DataFrame({
"timestamp": timestamps, "line": lines, "machine": machines,
"actual_anomaly": actual, "anomaly_score": score,
"criticality": criticality, "impact_k_yen": impact_k_yen,
"elapsed_min": elapsed_min,
})
df["date"] = df["timestamp"].dt.date
print(f"観測数: {len(df):,}件 / 真の異常: {df['actual_anomaly'].sum():,}件 "
f"({df['actual_anomaly'].mean():.1%})")
display(df.head())観測数: 1,200件 / 真の異常: 64件 (5.3%)| timestamp | line | machine | actual_anomaly | anomaly_score | criticality | impact_k_yen | elapsed_min | date | |
|---|---|---|---|---|---|---|---|---|---|
| 0 | 2025-01-01 00:00:00 | Line-B | Line-B-M2 | 0 | 0.253 | 2 | 962 | 10 | 2025-01-01 |
| 1 | 2025-01-01 00:36:00 | Line-C | Line-C-M2 | 0 | 0.367 | 1 | 622 | 98 | 2025-01-01 |
| 2 | 2025-01-01 01:12:00 | Line-A | Line-A-M2 | 1 | 0.610 | 3 | 1761 | 177 | 2025-01-01 |
| 3 | 2025-01-01 01:48:00 | Line-A | Line-A-M1 | 0 | 0.264 | 3 | 1059 | 25 | 2025-01-01 |
| 4 | 2025-01-01 02:24:00 | Line-A | Line-A-M2 | 0 | 0.113 | 3 | 1098 | 20 | 2025-01-01 |
fig, ax = plt.subplots(figsize=(9, 4.5))
ax.hist(df.loc[df["actual_anomaly"] == 0, "anomaly_score"], bins=25,
alpha=0.65, label="Normal", color="#4C78A8")
ax.hist(df.loc[df["actual_anomaly"] == 1, "anomaly_score"], bins=25,
alpha=0.65, label="Anomaly", color="#E45756")
ax.set_title("Anomaly score distribution by actual class")
ax.set_xlabel("Anomaly score")
ax.set_ylabel("Number of observations")
ax.grid(True, alpha=0.3)
ax.legend()
fig.tight_layout()
plt.show()
No.081：異常検知の評価指標を整理する
実務での意味
最初に、予測と実績を TP（正しく検知）、FP（誤検知）、FN（見逃し）、TN（正しく正常判定）の4区分へ分けます。工場では、FPは不要点検、FNは突発停止リスクに対応します。Accuracyだけでなく、この内訳を管理会議で共有することが重要です。
分析・モデル化の考え方
しきい値を とし、スコア をアラートとします。代表指標は次の通りです。
\mathrm{Recall}=\frac{TP}{TP+FN}$$ まず暫定しきい値を0.50として、評価表を作ります。 ### Pythonで確認する ```python def metrics_at_threshold(data, threshold): y_true = data["actual_anomaly"].to_numpy() y_pred = (data["anomaly_score"].to_numpy() >= threshold).astype(int) tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel() return {"threshold": threshold, "TP": tp, "FP": fp, "FN": fn, "TN": tn, "precision": precision_score(y_true, y_pred, zero_division=0), "recall": recall_score(y_true, y_pred, zero_division=0), "f1": f1_score(y_true, y_pred, zero_division=0), "accuracy": (tp + tn) / len(y_true)} base = metrics_at_threshold(df, 0.50) display(pd.DataFrame([base])) display(pd.DataFrame([[base["TN"], base["FP"]], [base["FN"], base["TP"]]], index=["Actual normal", "Actual anomaly"], columns=["Predicted normal", "Predicted alert"])) ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>threshold</th> <th>TP</th> <th>FP</th> <th>FN</th> <th>TN</th> <th>precision</th> <th>recall</th> <th>f1</th> <th>accuracy</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>0.500</td> <td>53</td> <td>41</td> <td>11</td> <td>1095</td> <td>0.564</td> <td>0.828</td> <td>0.671</td> <td>0.957</td> </tr> </tbody> </table> <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>Predicted normal</th> <th>Predicted alert</th> </tr> </thead> <tbody> <tr> <th>Actual normal</th> <td>1095</td> <td>41</td> </tr> <tr> <th>Actual anomaly</th> <td>11</td> <td>53</td> </tr> </tbody> </table> ### 結果の読み取り Accuracyが高くても、異常が希少なため安心はできません。TP・FP・FNを実件数で見ると、点検担当へ何件の作業を発生させ、何件の兆候を取り逃がすかが明確になります。以後の各指標は、この混同行列を異なる角度から要約したものです。 ## No.082：Precision・Recallを確認する ### 実務での意味 Precisionは「アラートを開いたら、どの程度が本当に異常か」であり、現場の信頼感と点検効率に関係します。Recallは「実際の異常をどの程度拾えたか」であり、保全上の安全網の強さに関係します。 ### 分析・モデル化の考え方 しきい値を下げると一般にRecallは上がり、Precisionは下がります。単一の値だけでなく、候補しきい値ごとの両指標とアラート件数を比較します。 ### Pythonで確認する ```python thresholds = np.arange(0.20, 0.86, 0.05) tradeoff = pd.DataFrame([metrics_at_threshold(df, t) for t in thresholds]) tradeoff["alerts"] = tradeoff["TP"] + tradeoff["FP"] display(tradeoff[["threshold", "precision", "recall", "alerts"]].round(3)) fig, ax = plt.subplots(figsize=(8, 4.5)) ax.plot(tradeoff["threshold"], tradeoff["precision"], marker="o", label="Precision") ax.plot(tradeoff["threshold"], tradeoff["recall"], marker="s", label="Recall") ax.set_title("Precision and recall by threshold") ax.set_xlabel("Threshold") ax.set_ylabel("Metric value") ax.set_ylim(0, 1.05) ax.grid(True, alpha=0.3) ax.legend() fig.tight_layout() plt.show() ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>threshold</th> <th>precision</th> <th>recall</th> <th>alerts</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>0.200</td> <td>0.114</td> <td>1.000</td> <td>561</td> </tr> <tr> <th>1</th> <td>0.250</td> <td>0.149</td> <td>0.984</td> <td>422</td> </tr> <tr> <th>2</th> <td>0.300</td> <td>0.194</td> <td>0.984</td> <td>325</td> </tr> <tr> <th>3</th> <td>0.350</td> <td>0.255</td> <td>0.938</td> <td>235</td> </tr> <tr> <th>4</th> <td>0.400</td> <td>0.355</td> <td>0.938</td> <td>169</td> </tr> <tr> <th>5</th> <td>0.450</td> <td>0.483</td> <td>0.906</td> <td>120</td> </tr> <tr> <th>6</th> <td>0.500</td> <td>0.564</td> <td>0.828</td> <td>94</td> </tr> <tr> <th>7</th> <td>0.550</td> <td>0.681</td> <td>0.766</td> <td>72</td> </tr> <tr> <th>8</th> <td>0.600</td> <td>0.833</td> <td>0.703</td> <td>54</td> </tr> <tr> <th>9</th> <td>0.650</td> <td>0.891</td> <td>0.641</td> <td>46</td> </tr> <tr> <th>10</th> <td>0.700</td> <td>0.900</td> <td>0.562</td> <td>40</td> </tr> <tr> <th>11</th> <td>0.750</td> <td>0.963</td> <td>0.406</td> <td>27</td> </tr> <tr> <th>12</th> <td>0.800</td> <td>1.000</td> <td>0.328</td> <td>21</td> </tr> <tr> <th>13</th> <td>0.850</td> <td>1.000</td> <td>0.156</td> <td>10</td> </tr> </tbody> </table>  ### 結果の読み取り 低いしきい値は見逃しを減らす一方、多数の誤検知を発生させます。高いしきい値は点検効率を改善しますが、異常を拾えません。工場では「Recall 90％以上」のような安全側制約と「1日10件以内」のような処理能力制約を併記すると、議論が具体化します。 ## No.083：F1スコアを確認する ### 実務での意味 複数モデルや複数しきい値を一つの数字で比較したいとき、PrecisionとRecallの調和平均であるF1が便利です。ただし、誤検知と見逃しを同じ重さで扱うため、最終的な業務判断をF1だけに委ねてはいけません。 ### 分析・モデル化の考え方 $$F_1=2\frac{\mathrm{Precision}\times\mathrm{Recall}} {\mathrm{Precision}+\mathrm{Recall}}$$ どちらか一方が低いとF1も低くなります。候補の中で最大となる点を確認します。 ### Pythonで確認する ```python f1_best = tradeoff.loc[tradeoff["f1"].idxmax()] display(tradeoff[["threshold", "precision", "recall", "f1"]].round(3)) print(f"候補中のF1最大: threshold={f1_best['threshold']:.2f}, " f"F1={f1_best['f1']:.3f}") fig, ax = plt.subplots(figsize=(8, 4.5)) ax.plot(tradeoff["threshold"], tradeoff["f1"], marker="o", color="#59A14F") ax.axvline(f1_best["threshold"], linestyle="--", color="#E15759", label="Best F1") ax.set_title("F1 score by threshold") ax.set_xlabel("Threshold") ax.set_ylabel("F1 score") ax.set_ylim(0, 1.05) ax.grid(True, alpha=0.3) ax.legend() fig.tight_layout() plt.show() ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>threshold</th> <th>precision</th> <th>recall</th> <th>f1</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>0.200</td> <td>0.114</td> <td>1.000</td> <td>0.205</td> </tr> <tr> <th>1</th> <td>0.250</td> <td>0.149</td> <td>0.984</td> <td>0.259</td> </tr> <tr> <th>2</th> <td>0.300</td> <td>0.194</td> <td>0.984</td> <td>0.324</td> </tr> <tr> <th>3</th> <td>0.350</td> <td>0.255</td> <td>0.938</td> <td>0.401</td> </tr> <tr> <th>4</th> <td>0.400</td> <td>0.355</td> <td>0.938</td> <td>0.515</td> </tr> <tr> <th>5</th> <td>0.450</td> <td>0.483</td> <td>0.906</td> <td>0.630</td> </tr> <tr> <th>6</th> <td>0.500</td> <td>0.564</td> <td>0.828</td> <td>0.671</td> </tr> <tr> <th>7</th> <td>0.550</td> <td>0.681</td> <td>0.766</td> <td>0.721</td> </tr> <tr> <th>8</th> <td>0.600</td> <td>0.833</td> <td>0.703</td> <td>0.763</td> </tr> <tr> <th>9</th> <td>0.650</td> <td>0.891</td> <td>0.641</td> <td>0.745</td> </tr> <tr> <th>10</th> <td>0.700</td> <td>0.900</td> <td>0.562</td> <td>0.692</td> </tr> <tr> <th>11</th> <td>0.750</td> <td>0.963</td> <td>0.406</td> <td>0.571</td> </tr> <tr> <th>12</th> <td>0.800</td> <td>1.000</td> <td>0.328</td> <td>0.494</td> </tr> <tr> <th>13</th> <td>0.850</td> <td>1.000</td> <td>0.156</td> <td>0.270</td> </tr> </tbody> </table> 候補中のF1最大: threshold=0.60, F1=0.763  ### 結果の読み取り F1最大点は統計的な基準線として有用です。ただし、停止損失が点検費より桁違いに大きい工場では、F1最大点より低いしきい値が合理的な場合があります。F1は「候補を絞る指標」、コスト評価は「運用点を決める指標」と役割を分けます。 ## No.084：ROC-AUCを確認する ### 実務での意味 ROC-AUCは、無作為に選んだ異常データが正常データより高いスコアになる確率として解釈できます。しきい値を固定する前に、モデルが異常を上位へ並べる能力を確認する指標です。 ### 分析・モデル化の考え方 ROC曲線は偽陽性率 $FPR=FP/(FP+TN)$ と真陽性率（Recall）を全しきい値で描きます。AUCが1に近いほど順位付けが良く、0.5はランダム相当です。ただし異常が希少な場合、多数のTNによりFPRが小さく見える点に注意します。 ### Pythonで確認する ```python y_true = df["actual_anomaly"].to_numpy() y_score = df["anomaly_score"].to_numpy() fpr, tpr, roc_thresholds = roc_curve(y_true, y_score) roc_auc = roc_auc_score(y_true, y_score) print(f"ROC-AUC: {roc_auc:.3f}") fig, ax = plt.subplots(figsize=(6, 5)) ax.plot(fpr, tpr, label=f"ROC-AUC = {roc_auc:.3f}", color="#4C78A8") ax.plot([0, 1], [0, 1], linestyle="--", color="gray", label="Random") ax.set_title("ROC curve") ax.set_xlabel("False positive rate") ax.set_ylabel("True positive rate") ax.grid(True, alpha=0.3) ax.legend() fig.tight_layout() plt.show() ``` ROC-AUC: 0.976  ### 結果の読み取り ROC-AUCが高ければ、スコアによる優先順位には利用価値があります。ただし「1日何件の誤検知になるか」はAUCから分かりません。クラス不均衡が強い異常検知では、次のPR-AUCと実件数を必ず併用します。 ## No.085：PR-AUCを確認する ### 実務での意味 PR曲線は、アラートの的中率（Precision）と異常の捕捉率（Recall）の交換関係を直接表します。異常が希少な製造データでは、ROC曲線より現場負荷を想像しやすい評価です。 ### 分析・モデル化の考え方 ここではPR曲線の要約としてAverage Precision（AP）を使います。ランダムな順位付けの基準はおおむね異常率です。したがってAPは絶対値だけでなく、ベースラインである異常率からどれだけ改善したかを見ます。 ### Pythonで確認する ```python precision_curve, recall_curve, pr_thresholds = precision_recall_curve(y_true, y_score) pr_auc = average_precision_score(y_true, y_score) baseline = y_true.mean() print(f"PR-AUC (Average Precision): {pr_auc:.3f}") print(f"ランダム基準（異常率） : {baseline:.3f}") fig, ax = plt.subplots(figsize=(6, 5)) ax.plot(recall_curve, precision_curve, label=f"AP = {pr_auc:.3f}", color="#F28E2B") ax.axhline(baseline, linestyle="--", color="gray", label=f"Baseline = {baseline:.3f}") ax.set_title("Precision-recall curve") ax.set_xlabel("Recall") ax.set_ylabel("Precision") ax.set_xlim(0, 1.02) ax.set_ylim(0, 1.02) ax.grid(True, alpha=0.3) ax.legend() fig.tight_layout() plt.show() ``` PR-AUC (Average Precision): 0.835 ランダム基準（異常率） : 0.053  ### 結果の読み取り PR-AUCが異常率を大きく上回るなら、アラート上位を点検する運用には意味があります。ただしAPも平均値です。現場の処理能力が上位50件なら、その範囲でのPrecisionや捕捉件数を別途確認します。 ## No.086：誤検知と見逃しのコストを整理する ### 実務での意味 同じ1件でも、FPは点検時間、FNは停止・廃棄・納期遅延を生みます。誤差を金額へ変換すると、製造、保全、経営が同じ尺度でしきい値を議論できます。 ### 分析・モデル化の考え方 簡単な期待コストを次で定義します。 $$C(t)=c_{FP}\,FP(t)+c_{FN}\,FN(t)+c_{TP}\,TP(t)$$ $c_{FP}$ は不要点検費、$c_{FN}$ は見逃し損失、$c_{TP}$ は正しく検知した際の予防点検費です。ここでは 1件あたりそれぞれ2万円、80万円、5万円と仮定します。実務では設備別の停止影響額や安全影響で細分化します。 ### Pythonで確認する ```python COST_FP = 20_000 COST_FN = 800_000 COST_TP = 50_000 cost_table = tradeoff.copy() cost_table["false_alarm_cost_yen"] = cost_table["FP"] * COST_FP cost_table["miss_cost_yen"] = cost_table["FN"] * COST_FN cost_table["planned_check_cost_yen"] = cost_table["TP"] * COST_TP cost_table["total_cost_yen"] = cost_table[[ "false_alarm_cost_yen", "miss_cost_yen", "planned_check_cost_yen" ]].sum(axis=1) display(cost_table[["threshold", "FP", "FN", "TP", "total_cost_yen"]] .assign(total_cost_yen=lambda x: x["total_cost_yen"].map("¥{:,.0f}".format))) ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>threshold</th> <th>FP</th> <th>FN</th> <th>TP</th> <th>total_cost_yen</th> </tr> </thead> <tbody> <tr> <th>0</th> <td>0.200</td> <td>497</td> <td>0</td> <td>64</td> <td>¥13,140,000</td> </tr> <tr> <th>1</th> <td>0.250</td> <td>359</td> <td>1</td> <td>63</td> <td>¥11,130,000</td> </tr> <tr> <th>2</th> <td>0.300</td> <td>262</td> <td>1</td> <td>63</td> <td>¥9,190,000</td> </tr> <tr> <th>3</th> <td>0.350</td> <td>175</td> <td>4</td> <td>60</td> <td>¥9,700,000</td> </tr> <tr> <th>4</th> <td>0.400</td> <td>109</td> <td>4</td> <td>60</td> <td>¥8,380,000</td> </tr> <tr> <th>5</th> <td>0.450</td> <td>62</td> <td>6</td> <td>58</td> <td>¥8,940,000</td> </tr> <tr> <th>6</th> <td>0.500</td> <td>41</td> <td>11</td> <td>53</td> <td>¥12,270,000</td> </tr> <tr> <th>7</th> <td>0.550</td> <td>23</td> <td>15</td> <td>49</td> <td>¥14,910,000</td> </tr> <tr> <th>8</th> <td>0.600</td> <td>9</td> <td>19</td> <td>45</td> <td>¥17,630,000</td> </tr> <tr> <th>9</th> <td>0.650</td> <td>5</td> <td>23</td> <td>41</td> <td>¥20,550,000</td> </tr> <tr> <th>10</th> <td>0.700</td> <td>4</td> <td>28</td> <td>36</td> <td>¥24,280,000</td> </tr> <tr> <th>11</th> <td>0.750</td> <td>1</td> <td>38</td> <td>26</td> <td>¥31,720,000</td> </tr> <tr> <th>12</th> <td>0.800</td> <td>0</td> <td>43</td> <td>21</td> <td>¥35,450,000</td> </tr> <tr> <th>13</th> <td>0.850</td> <td>0</td> <td>54</td> <td>10</td> <td>¥43,700,000</td> </tr> </tbody> </table> ### 結果の読み取り 見逃し単価を高く置いたため、少数のFNでも総コストへの影響が大きくなります。単価は精密な会計値でなくても、悲観・標準・楽観の3シナリオを置けば意思決定の頑健性を確認できます。安全・法令に関わる異常は金額換算だけでなく「Recall下限」の制約として扱います。 ## No.087：しきい値を業務コストから決める ### 実務での意味 しきい値はモデル担当者だけで決めるパラメータではありません。保全の処理能力、停止損失、安全要件を反映した業務ルールです。最小コスト点と運用制約を同時に示し、承認可能な根拠を作ります。 ### 分析・モデル化の考え方 候補しきい値を細かく走査し、期待コストを最小にします。さらにRecall 85％以上を必須条件とする例も確認します。制約付き最適化にすることで、安全要件をコストで相殺しない設計になります。 ### Pythonで確認する ```python fine_thresholds = np.linspace(0.10, 0.90, 161) optimization = pd.DataFrame([metrics_at_threshold(df, t) for t in fine_thresholds]) optimization["alerts"] = optimization["TP"] + optimization["FP"] optimization["total_cost_yen"] = ( optimization["FP"] * COST_FP + optimization["FN"] * COST_FN + optimization["TP"] * COST_TP ) best_cost = optimization.loc[optimization["total_cost_yen"].idxmin()] feasible = optimization[optimization["recall"] >= 0.85] best_constrained = feasible.loc[feasible["total_cost_yen"].idxmin()] display(pd.DataFrame([best_cost, best_constrained], index=["Minimum cost", "Minimum cost with Recall >= 0.85"])[ ["threshold", "precision", "recall", "f1", "alerts", "total_cost_yen"] ].round(3)) fig, ax = plt.subplots(figsize=(8, 4.5)) ax.plot(optimization["threshold"], optimization["total_cost_yen"] / 1_000_000, color="#B279A2") ax.axvline(best_constrained["threshold"], linestyle="--", color="#E15759", label="Selected threshold") ax.set_title("Expected business cost by threshold") ax.set_xlabel("Threshold") ax.set_ylabel("Total cost (million JPY)") ax.grid(True, alpha=0.3) ax.legend() fig.tight_layout() plt.show() SELECTED_THRESHOLD = float(best_constrained["threshold"]) df["alert"] = (df["anomaly_score"] >= SELECTED_THRESHOLD).astype(int) print(f"採用しきい値: {SELECTED_THRESHOLD:.3f}") ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>threshold</th> <th>precision</th> <th>recall</th> <th>f1</th> <th>alerts</th> <th>total_cost_yen</th> </tr> </thead> <tbody> <tr> <th>Minimum cost</th> <td>0.440</td> <td>0.451</td> <td>0.938</td> <td>0.609</td> <td>133.000</td> <td>7,660,000.000</td> </tr> <tr> <th>Minimum cost with Recall >= 0.85</th> <td>0.440</td> <td>0.451</td> <td>0.938</td> <td>0.609</td> <td>133.000</td> <td>7,660,000.000</td> </tr> </tbody> </table>  採用しきい値: 0.440 ### 結果の読み取り 採用値は「F1最大」ではなく「Recall制約を満たす中で期待コスト最小」という説明可能なルールで決まりました。単価や制約が変われば最適値も変わるため、しきい値と前提条件をセットで版管理します。 ## No.088：アラート件数を集計する ### 実務での意味 総件数が許容範囲でも、特定の日やラインへ集中すれば対応不能になります。日別・ライン別の負荷を可視化し、シフト人数や巡回計画へ接続します。 ### 分析・モデル化の考え方 アラートを日付とラインで集計し、日次処理能力を超える日を検出します。ここでは工場全体で1日8件を標準処理能力と仮定します。集計粒度は、実運用の担当割当単位に合わせます。 ### Pythonで確認する ```python daily_line = (df[df["alert"] == 1] .pivot_table(index="date", columns="line", values="alert", aggfunc="sum", fill_value=0)) daily_line["Total"] = daily_line.sum(axis=1) DAILY_CAPACITY = 8 daily_line["Over capacity"] = daily_line["Total"] > DAILY_CAPACITY display(daily_line.head(10)) print(f"処理能力超過日: {daily_line['Over capacity'].sum()}日 / {len(daily_line)}日") fig, ax = plt.subplots(figsize=(10, 4.8)) plot_cols = [c for c in ["Line-A", "Line-B", "Line-C"] if c in daily_line.columns] daily_line[plot_cols].plot(kind="bar", stacked=True, ax=ax, color=["#4C78A8", "#F28E2B", "#59A14F"][:len(plot_cols)]) ax.axhline(DAILY_CAPACITY, linestyle="--", color="#E15759", label="Daily capacity") ax.set_title("Daily alerts by production line") ax.set_xlabel("Date") ax.set_ylabel("Number of alerts") ax.grid(True, axis="y", alpha=0.3) ax.legend() ax.tick_params(axis="x", labelrotation=75) fig.tight_layout() plt.show() ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th>line</th> <th>Line-A</th> <th>Line-B</th> <th>Line-C</th> <th>Total</th> <th>Over capacity</th> </tr> <tr> <th>date</th> <th></th> <th></th> <th></th> <th></th> <th></th> </tr> </thead> <tbody> <tr> <th>2025-01-01</th> <td>3</td> <td>4</td> <td>2</td> <td>9</td> <td>True</td> </tr> <tr> <th>2025-01-02</th> <td>1</td> <td>1</td> <td>0</td> <td>2</td> <td>False</td> </tr> <tr> <th>2025-01-03</th> <td>0</td> <td>2</td> <td>0</td> <td>2</td> <td>False</td> </tr> <tr> <th>2025-01-04</th> <td>3</td> <td>0</td> <td>1</td> <td>4</td> <td>False</td> </tr> <tr> <th>2025-01-05</th> <td>2</td> <td>0</td> <td>2</td> <td>4</td> <td>False</td> </tr> <tr> <th>2025-01-06</th> <td>3</td> <td>5</td> <td>3</td> <td>11</td> <td>True</td> </tr> <tr> <th>2025-01-07</th> <td>4</td> <td>2</td> <td>1</td> <td>7</td> <td>False</td> </tr> <tr> <th>2025-01-08</th> <td>0</td> <td>4</td> <td>0</td> <td>4</td> <td>False</td> </tr> <tr> <th>2025-01-09</th> <td>4</td> <td>1</td> <td>1</td> <td>6</td> <td>False</td> </tr> <tr> <th>2025-01-10</th> <td>2</td> <td>0</td> <td>1</td> <td>3</td> <td>False</td> </tr> </tbody> </table> 処理能力超過日: 3日 / 30日  ### 結果の読み取り 処理能力超過日があれば、単純にしきい値を上げる前に、同一設備の重複抑制、シフト別の再配分、重要度によるキュー分割を検討します。件数KPIは平均だけでなく、最大値や95％点、超過日数を監視します。 ## No.089：アラートの優先度を設計する ### 実務での意味 全アラートを同列に扱うと、重要設備の兆候が軽微なアラートの後ろに埋もれます。異常らしさ、停止影響、緊急度を統合し、限られた担当者が上から処理できる順位を作ります。 ### 分析・モデル化の考え方 例として、各要素を0〜1へそろえ、次の優先度スコアを定義します。 $$P=0.50\,S+0.30\,I+0.20\,U$$ $S$ は異常スコア、$I$ は設備重要度、$U$ は経過時間に基づく緊急度です。重みは仮定であり、過去の対応結果と現場レビューで更新します。 ### Pythonで確認する ```python alerts = df[df["alert"] == 1].copy() alerts["importance_norm"] = alerts["criticality"] / 3 alerts["urgency_norm"] = np.clip(alerts["elapsed_min"] / 120, 0, 1) alerts["priority_score"] = ( 0.50 * alerts["anomaly_score"] + 0.30 * alerts["importance_norm"] + 0.20 * alerts["urgency_norm"] ) alerts["priority"] = pd.cut(alerts["priority_score"], bins=[-np.inf, 0.65, 0.80, np.inf], labels=["P3", "P2", "P1"]) priority_counts = alerts["priority"].value_counts().reindex(["P1", "P2", "P3"], fill_value=0) display(priority_counts.rename("alerts").to_frame()) display(alerts.sort_values("priority_score", ascending=False)[[ "timestamp", "line", "machine", "anomaly_score", "criticality", "impact_k_yen", "elapsed_min", "priority_score", "priority" ]].head(10)) fig, ax = plt.subplots(figsize=(6.5, 4.2)) priority_counts.plot(kind="bar", ax=ax, color=["#E15759", "#F28E2B", "#4C78A8"]) ax.set_title("Alert count by priority") ax.set_xlabel("Priority") ax.set_ylabel("Number of alerts") ax.grid(True, axis="y", alpha=0.3) ax.tick_params(axis="x", labelrotation=0) fig.tight_layout() plt.show() ``` <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>alerts</th> </tr> <tr> <th>priority</th> <th></th> </tr> </thead> <tbody> <tr> <th>P1</th> <td>19</td> </tr> <tr> <th>P2</th> <td>55</td> </tr> <tr> <th>P3</th> <td>59</td> </tr> </tbody> </table> <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>timestamp</th> <th>line</th> <th>machine</th> <th>anomaly_score</th> <th>criticality</th> <th>impact_k_yen</th> <th>elapsed_min</th> <th>priority_score</th> <th>priority</th> </tr> </thead> <tbody> <tr> <th>786</th> <td>2025-01-20 15:36:00</td> <td>Line-A</td> <td>Line-A-M2</td> <td>0.928</td> <td>3</td> <td>1587</td> <td>121</td> <td>0.964</td> <td>P1</td> </tr> <tr> <th>716</th> <td>2025-01-18 21:36:00</td> <td>Line-A</td> <td>Line-A-M1</td> <td>0.896</td> <td>3</td> <td>1473</td> <td>150</td> <td>0.948</td> <td>P1</td> </tr> <tr> <th>237</th> <td>2025-01-06 22:12:00</td> <td>Line-A</td> <td>Line-A-M2</td> <td>0.843</td> <td>3</td> <td>1296</td> <td>166</td> <td>0.922</td> <td>P1</td> </tr> <tr> <th>735</th> <td>2025-01-19 09:00:00</td> <td>Line-A</td> <td>Line-A-M2</td> <td>0.803</td> <td>3</td> <td>1341</td> <td>132</td> <td>0.901</td> <td>P1</td> </tr> <tr> <th>574</th> <td>2025-01-15 08:24:00</td> <td>Line-A</td> <td>Line-A-M1</td> <td>0.898</td> <td>3</td> <td>1587</td> <td>88</td> <td>0.896</td> <td>P1</td> </tr> <tr> <th>1083</th> <td>2025-01-28 01:48:00</td> <td>Line-A</td> <td>Line-A-M1</td> <td>0.855</td> <td>3</td> <td>1911</td> <td>92</td> <td>0.881</td> <td>P1</td> </tr> <tr> <th>266</th> <td>2025-01-07 15:36:00</td> <td>Line-A</td> <td>Line-A-M2</td> <td>0.776</td> <td>3</td> <td>1755</td> <td>112</td> <td>0.874</td> <td>P1</td> </tr> <tr> <th>751</th> <td>2025-01-19 18:36:00</td> <td>Line-A</td> <td>Line-A-M2</td> <td>0.727</td> <td>3</td> <td>1899</td> <td>143</td> <td>0.864</td> <td>P1</td> </tr> <tr> <th>249</th> <td>2025-01-07 05:24:00</td> <td>Line-A</td> <td>Line-A-M1</td> <td>0.722</td> <td>3</td> <td>1845</td> <td>137</td> <td>0.861</td> <td>P1</td> </tr> <tr> <th>1162</th> <td>2025-01-30 01:12:00</td> <td>Line-A</td> <td>Line-A-M1</td> <td>0.827</td> <td>3</td> <td>1899</td> <td>88</td> <td>0.860</td> <td>P1</td> </tr> </tbody> </table>  ### 結果の読み取り 優先度は「モデルスコアの言い換え」ではなく、事業影響を加えた対応順です。P1が常に多すぎる場合は区分が機能していません。P1は即時対応可能な件数に抑え、各要素と重みを監査できるように残します。 ## No.090：現場で使えるアラート設計を考える ### 実務での意味 最後に、判定から終結までの責任と期限を決めます。モデルが良くても、通知先、重複抑制、確認手順、結果記録がなければ改善サイクルは回りません。 ### 分析・モデル化の考え方 優先度ごとにSLAと通知先を割り当てます。また、同一設備の近接アラートは1件にまとめる「クールダウン」を設けます。ここでは同一設備で120分以内に連続した通知を抑制し、担当者が見る運用キューを作ります。 ### Pythonで確認する ```python alerts = alerts.sort_values(["machine", "timestamp"]).copy() alerts["minutes_since_previous"] = ( alerts.groupby("machine")["timestamp"].diff().dt.total_seconds().div(60) ) COOLDOWN_MIN = 120 alerts["notify"] = alerts["minutes_since_previous"].isna() | ( alerts["minutes_since_previous"] > COOLDOWN_MIN ) sla_map = {"P1": 15, "P2": 60, "P3": 240} owner_map = {"P1": "保全責任者＋製造課長", "P2": "当番保全員", "P3": "日次巡回担当"} alerts["sla_min"] = alerts["priority"].astype(str).map(sla_map) alerts["owner"] = alerts["priority"].astype(str).map(owner_map) alerts["due_at"] = alerts["timestamp"] + pd.to_timedelta(alerts["sla_min"], unit="m") queue = alerts[alerts["notify"]].sort_values( ["priority_score", "timestamp"], ascending=[False, True] ) print(f"抑制前: {len(alerts)}件 / 通知後: {len(queue)}件 / " f"抑制率: {1 - len(queue) / len(alerts):.1%}") display(queue[["timestamp", "machine", "priority", "owner", "sla_min", "due_at", "anomaly_score", "priority_score"]].head(12)) ``` 抑制前: 133件 / 通知後: 125件 / 抑制率: 6.0% <style scoped> .dataframe tbody tr th:only-of-type { vertical-align: middle; } .dataframe tbody tr th { vertical-align: top; } .dataframe thead th { text-align: right; } </style> <table border="1" class="dataframe"> <thead> <tr style="text-align: right;"> <th></th> <th>timestamp</th> <th>machine</th> <th>priority</th> <th>owner</th> <th>sla_min</th> <th>due_at</th> <th>anomaly_score</th> <th>priority_score</th> </tr> </thead> <tbody> <tr> <th>786</th> <td>2025-01-20 15:36:00</td> <td>Line-A-M2</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-20 15:51:00</td> <td>0.928</td> <td>0.964</td> </tr> <tr> <th>716</th> <td>2025-01-18 21:36:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-18 21:51:00</td> <td>0.896</td> <td>0.948</td> </tr> <tr> <th>237</th> <td>2025-01-06 22:12:00</td> <td>Line-A-M2</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-06 22:27:00</td> <td>0.843</td> <td>0.922</td> </tr> <tr> <th>574</th> <td>2025-01-15 08:24:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-15 08:39:00</td> <td>0.898</td> <td>0.896</td> </tr> <tr> <th>1083</th> <td>2025-01-28 01:48:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-28 02:03:00</td> <td>0.855</td> <td>0.881</td> </tr> <tr> <th>266</th> <td>2025-01-07 15:36:00</td> <td>Line-A-M2</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-07 15:51:00</td> <td>0.776</td> <td>0.874</td> </tr> <tr> <th>751</th> <td>2025-01-19 18:36:00</td> <td>Line-A-M2</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-19 18:51:00</td> <td>0.727</td> <td>0.864</td> </tr> <tr> <th>249</th> <td>2025-01-07 05:24:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-07 05:39:00</td> <td>0.722</td> <td>0.861</td> </tr> <tr> <th>1162</th> <td>2025-01-30 01:12:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-30 01:27:00</td> <td>0.827</td> <td>0.860</td> </tr> <tr> <th>1012</th> <td>2025-01-26 07:12:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-26 07:27:00</td> <td>0.862</td> <td>0.856</td> </tr> <tr> <th>975</th> <td>2025-01-25 09:00:00</td> <td>Line-A-M1</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-25 09:15:00</td> <td>0.724</td> <td>0.844</td> </tr> <tr> <th>1175</th> <td>2025-01-30 09:00:00</td> <td>Line-B-M2</td> <td>P1</td> <td>保全責任者＋製造課長</td> <td>15</td> <td>2025-01-30 09:15:00</td> <td>0.864</td> <td>0.832</td> </tr> </tbody> </table> ### 結果の読み取り 通知抑制により担当者が見る件数を減らしつつ、P1から処理するキューを作れました。本番では「確認済み」「要観察」「作業指示」「異常なし」「終結」の状態遷移、未対応時のエスカレーション、判断理由の記録も必要です。抑制された通知も監査用には消さず、元イベントとの対応を保存します。 ## 対象ノックを通して見える実務上の示唆 1. **モデル性能と運用品質は別物**：AUCが高くても、通知集中や重複で現場が処理できなければ価値は出ません。 2. **しきい値は経営条件で変わる**：見逃し損失、点検費、安全制約、要員数を変えると採用値も変わります。 3. **希少事象では実件数を併記する**：割合だけでなく、TP・FP・FN、日次アラート数、能力超過日を示します。 4. **優先順位には事業影響を入れる**：異常スコアだけでなく、重要設備、停止影響、経過時間を統合します。 5. **対応結果が次の教師データになる**：現場の判定と処置を構造化して残すことで、再評価と改善が可能になります。 ## 実務導入する場合に必要なこと - 正解ラベルの定義：故障、品質異常、兆候、誤報の判定基準と確定期限 - コスト合意：誤検知工数、停止時間、廃棄、納期・安全影響の評価 - 運用能力：シフト別の担当者数、優先度別SLA、夜間休日の連絡網 - システム連携：センサー基盤、保全管理システム、通知、監査ログの接続 - KPI：Recall、Precision、FN件数、対応時間、能力超過日、抑制率、再発率 - ガバナンス：しきい値変更承認、モデル・データの版管理、定期レビュー 導入初期は、アラートを作業指示へ自動変換せず、シャドー運用で件数と妥当性を確認するのが現実的です。設備群ごとに4〜8週間の試行を行い、現場レビューを経て段階的に対象を広げます。 ## まとめ No.081〜No.090では、混同行列からPrecision・Recall・F1、ROC-AUC、PR-AUCへ進み、誤検知と見逃しを業務コストへ翻訳しました。さらに、コストとRecall制約からしきい値を選び、日別負荷、優先度、SLA、重複抑制まで設計しました。 異常検知の成果物はモデルファイルだけではありません。**評価基準、しきい値の根拠、アラートキュー、対応ルール、改善用ログを一体で設計すること**が、現場で使われ続ける仕組みにつながります。 ## 法人向けのご相談 数理工房では、異常検知PoC、評価設計、設備別の損失評価、しきい値最適化、アラート運用設計まで、製造現場の制約に合わせて支援します。既存モデルの誤検知が多い、評価指標を経営判断へ結びつけたい、保全システムと連携したい、といった段階からご相談いただけます。 > 📩 **お問い合わせ**: [surikobo.co.jp/contact](https://surikobo.co.jp/contact) > まずはお気軽にご相談ください。