feat: Tier 1 이상 탐지 SECOM 기반 구현

agents/detection.py

@@ -1,11 +1,63 @@
-"""Tier 1 이상 탐지 에이전트 (M2에서 구현)
+"""Tier 1 이상 탐지 에이전트
 
-입력: 알람 컨텍스트 + SECOM 센서 데이터 (data.secom.loader.load_secom)
-출력: core.schema.Tier1 (이상 점수 + 기여 피처 Top-N + 영향 lot)
-모델: GPT-5 mini + tool use, IsolationForest 등으로 이상 점수 계산
+알람을 SECOM 웨이퍼 row에 매핑해 이상 점수와 기여 피처를 계산
+모델: IsolationForest (experiments/tier1_detection 벤치마크에서 PR-AUC 가장 우수)
+이상 점수는 학습 분포 대비 백분위로 0~1 정규화
+기여 피처는 표준화 값의 절대크기 Top-N (정상 분포에서 가장 벗어난 센서)
 """
+from functools import lru_cache
+
+import numpy as np
+from sklearn.ensemble import IsolationForest
+
 from core.schema import Tier1
+from data.secom.loader import load_secom
+from data.secom.preprocess import SecomPreprocessor
+
+RANDOM_STATE = 42
+TOP_N_FEATURES = 3
+
+# SECOM은 익명화 데이터라 lot/step이 없음, MVP에선 알람을 특정 웨이퍼 row에 수동 매핑
+# secom_row: fail 라벨 row 인덱스, wafers: 데모 컨텍스트상 영향 웨이퍼 수
+ALARM_WAFER = {
+    "A1": {"secom_row": 2, "wafers": 25},
+}
+
+
+@lru_cache(maxsize=1)
+def _fit_model():
+    """SECOM 전체로 전처리기와 IsolationForest를 학습, 첫 호출 시 1회만"""
+    X, _ = load_secom()
+    pre = SecomPreprocessor().fit(X)
+    Xz = pre.transform(X)
+    model = IsolationForest(n_estimators=200, random_state=RANDOM_STATE)
+    model.fit(Xz)
+    train_scores = -model.score_samples(Xz)
+    return X, pre, model, train_scores
 
 
 def run_detection(alarm: dict) -> Tier1:
-    raise NotImplementedError("M2: Tier 1 이상 탐지 에이전트 구현 예정")
+    mapping = ALARM_WAFER.get(alarm["id"])
+    if mapping is None:
+        raise ValueError(f"SECOM 매핑이 없는 알람: {alarm['id']}")
+
+    X, pre, model, train_scores = _fit_model()
+    Xz = pre.transform(X.iloc[[mapping["secom_row"]]])
+
+    raw_score = float(-model.score_samples(Xz)[0])
+    # 학습 분포 대비 백분위, 높을수록 이상
+    score = float((train_scores < raw_score).mean())
+
+    # 표준화 값의 절대크기 = 정상 분포에서 벗어난 정도, Top-N
+    deviations = np.abs(Xz[0])
+    top_idx = np.argsort(deviations)[::-1][:TOP_N_FEATURES]
+    features = [
+        {"name": pre.keep_cols[i], "value": round(float(deviations[i]), 2)}
+        for i in top_idx
+    ]
+
+    return {
+        "score": round(score, 2),
+        "features": features,
+        "lot": {"id": alarm["lot_id"], "wafers": mapping["wafers"]},
+    }

data/secom/preprocess.py

@@ -0,0 +1,42 @@
+"""SECOM 전처리
+
+이상 탐지 모델에 넣기 전 공통 전처리
+- 전부 결측이거나 분산이 0(상수)인 컬럼 제거
+- 남은 결측치는 중앙값으로 임퓨테이션
+- StandardScaler로 스케일링
+
+agents/detection.py와 experiments/ 양쪽에서 공용으로 사용
+fit은 train 데이터에만, transform은 train/test 공통으로 적용
+"""
+import numpy as np
+import pandas as pd
+from sklearn.impute import SimpleImputer
+from sklearn.preprocessing import StandardScaler
+
+
+class SecomPreprocessor:
+    """SECOM 센서 데이터 전처리, sklearn 스타일 fit/transform"""
+
+    def __init__(self, var_threshold: float = 0.0):
+        self.var_threshold = var_threshold
+        self.keep_cols: list[str] = []
+        self.imputer = SimpleImputer(strategy="median")
+        self.scaler = StandardScaler()
+
+    def fit(self, X: pd.DataFrame) -> "SecomPreprocessor":
+        # 전부 결측인 컬럼 제거 후, 분산이 임계 이하(거의 상수)인 컬럼도 제거
+        non_empty = X.columns[X.notna().any()]
+        variances = X[non_empty].var()
+        self.keep_cols = list(variances[variances > self.var_threshold].index)
+
+        kept = X[self.keep_cols]
+        self.imputer.fit(kept)
+        self.scaler.fit(self.imputer.transform(kept))
+        return self
+
+    def transform(self, X: pd.DataFrame) -> np.ndarray:
+        kept = X[self.keep_cols]
+        return self.scaler.transform(self.imputer.transform(kept))
+
+    def fit_transform(self, X: pd.DataFrame) -> np.ndarray:
+        return self.fit(X).transform(X)

experiments/tier1_detection/benchmark.py

@@ -0,0 +1,147 @@
+"""Tier 1 이상 탐지 모델 벤치마크
+
+SECOM pass/fail 라벨을 정답으로 보고 비지도 이상 탐지 모델을 비교
+- IsolationForest
+- LocalOutlierFactor
+- OneClassSVM
+- baseline: 표준화 피처 절대값 평균 (단순 통계)
+
+평가: ROC-AUC, PR-AUC (fail이 6.6%뿐인 불균형 데이터라 PR-AUC를 주지표로 봄)
+train에만 fit, test에서 평가 (70/30 stratified split)
+
+실행: python -m experiments.tier1_detection.benchmark
+결과: results.md 표 + plots/ 그래프
+"""
+from pathlib import Path
+
+import matplotlib
+
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+from sklearn.ensemble import IsolationForest
+from sklearn.metrics import (
+    average_precision_score,
+    precision_recall_curve,
+    roc_auc_score,
+    roc_curve,
+)
+from sklearn.model_selection import train_test_split
+from sklearn.neighbors import LocalOutlierFactor
+from sklearn.svm import OneClassSVM
+
+from data.secom.loader import load_secom
+from data.secom.preprocess import SecomPreprocessor
+
+RANDOM_STATE = 42
+OUT_DIR = Path(__file__).parent
+PLOTS_DIR = OUT_DIR / "plots"
+
+MODELS = {
+    "IsolationForest": IsolationForest(n_estimators=200, random_state=RANDOM_STATE),
+    "LocalOutlierFactor": LocalOutlierFactor(n_neighbors=20, novelty=True),
+    "OneClassSVM": OneClassSVM(nu=0.1, kernel="rbf", gamma="scale"),
+    "baseline": None,
+}
+
+
+def anomaly_scores(name, model, X_train, X_test):
+    """모델별 이상 점수 반환, 높을수록 이상"""
+    if name == "baseline":
+        # 표준화된 피처의 절대값 평균, 정상에서 멀수록 큼
+        return np.abs(X_test).mean(axis=1)
+    model.fit(X_train)
+    # score_samples는 높을수록 정상이므로 부호 반전
+    return -model.score_samples(X_test)
+
+
+def main():
+    X, y = load_secom()
+    y_true = (y == 1).astype(int).to_numpy()  # 1 = fail(이상)
+
+    X_train_df, X_test_df, y_train, y_test = train_test_split(
+        X, y_true, test_size=0.3, stratify=y_true, random_state=RANDOM_STATE
+    )
+
+    pre = SecomPreprocessor().fit(X_train_df)
+    X_train = pre.transform(X_train_df)
+    X_test = pre.transform(X_test_df)
+
+    PLOTS_DIR.mkdir(exist_ok=True)
+    fig_roc, ax_roc = plt.subplots(figsize=(6, 5))
+    fig_pr, ax_pr = plt.subplots(figsize=(6, 5))
+
+    results = []
+    for name, model in MODELS.items():
+        scores = anomaly_scores(name, model, X_train, X_test)
+        roc = roc_auc_score(y_test, scores)
+        ap = average_precision_score(y_test, scores)
+        results.append((name, roc, ap))
+
+        fpr, tpr, _ = roc_curve(y_test, scores)
+        ax_roc.plot(fpr, tpr, label=f"{name} (AUC={roc:.3f})")
+        prec, rec, _ = precision_recall_curve(y_test, scores)
+        ax_pr.plot(rec, prec, label=f"{name} (AP={ap:.3f})")
+
+    # 참조선, plot 텍스트는 폰트 의존 피하려고 영문 사용
+    ax_roc.plot([0, 1], [0, 1], "k--", alpha=0.4, label="random")
+    ax_roc.set_xlabel("False Positive Rate")
+    ax_roc.set_ylabel("True Positive Rate")
+    ax_roc.set_title("Tier 1 Anomaly Detection - ROC")
+    ax_roc.legend(loc="lower right", fontsize=8)
+
+    ax_pr.axhline(y_test.mean(), color="k", ls="--", alpha=0.4, label="random")
+    ax_pr.set_xlabel("Recall")
+    ax_pr.set_ylabel("Precision")
+    ax_pr.set_title("Tier 1 Anomaly Detection - Precision-Recall")
+    ax_pr.legend(loc="upper right", fontsize=8)
+
+    fig_roc.tight_layout()
+    fig_roc.savefig(PLOTS_DIR / "roc.png", dpi=120)
+    fig_pr.tight_layout()
+    fig_pr.savefig(PLOTS_DIR / "pr.png", dpi=120)
+
+    write_results(results, len(X_train_df), len(X_test_df), int(y_test.sum()))
+    for name, roc, ap in sorted(results, key=lambda r: -r[2]):
+        print(f"{name:24s}  ROC-AUC={roc:.3f}  PR-AUC={ap:.3f}")
+
+
+def write_results(results, n_train, n_test, n_test_fail):
+    """results.md에 평가 표 기록, PR-AUC 내림차순 정렬"""
+    ranked = sorted(results, key=lambda r: -r[2])
+    lines = [
+        "# Tier 1 이상 탐지 - 모델 벤치마크",
+        "",
+        "SECOM 데이터셋(반도체 제조 공정 센서)으로 비지도 이상 탐지 모델을 비교합니다.",
+        "",
+        "## 설정",
+        "",
+        f"- train {n_train} / test {n_test} (70/30 stratified split)",
+        f"- test의 fail(이상) 샘플: {n_test_fail}건",
+        "- 전처리: 전결측/상수 컬럼 제거 -> 중앙값 임퓨테이션 -> 표준화",
+        "- 평가 지표: ROC-AUC, PR-AUC (불균형 데이터라 PR-AUC가 주지표)",
+        "",
+        "## 결과 (PR-AUC 내림차순)",
+        "",
+        "| 모델 | ROC-AUC | PR-AUC |",
+        "|---|---|---|",
+    ]
+    for name, roc, ap in ranked:
+        lines.append(f"| {name} | {roc:.3f} | {ap:.3f} |")
+    lines += [
+        "",
+        "![ROC](plots/roc.png)",
+        "",
+        "![PR](plots/pr.png)",
+        "",
+        f"## 채택",
+        "",
+        f"PR-AUC 기준 최고 모델은 **{ranked[0][0]}** "
+        f"(PR-AUC {ranked[0][2]:.3f}), agents/detection.py의 baseline으로 사용합니다.",
+        "",
+    ]
+    (OUT_DIR / "results.md").write_text("\n".join(lines), encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()

experiments/tier1_detection/results.md

@@ -0,0 +1,27 @@
+# Tier 1 이상 탐지 - 모델 벤치마크
+
+SECOM 데이터셋(반도체 제조 공정 센서)으로 비지도 이상 탐지 모델을 비교합니다.
+
+## 설정
+
+- train 1096 / test 471 (70/30 stratified split)
+- test의 fail(이상) 샘플: 31건
+- 전처리: 전결측/상수 컬럼 제거 -> 중앙값 임퓨테이션 -> 표준화
+- 평가 지표: ROC-AUC, PR-AUC (불균형 데이터라 PR-AUC가 주지표)
+
+## 결과 (PR-AUC 내림차순)
+
+| 모델 | ROC-AUC | PR-AUC |
+|---|---|---|
+| IsolationForest | 0.600 | 0.129 |
+| baseline | 0.565 | 0.119 |
+| OneClassSVM | 0.563 | 0.098 |
+| LocalOutlierFactor | 0.544 | 0.089 |
+
+![ROC](plots/roc.png)
+
+![PR](plots/pr.png)
+
+## 채택
+
+PR-AUC 기준 최고 모델은 **IsolationForest** (PR-AUC 0.129), agents/detection.py의 baseline으로 사용합니다.

requirements.txt

@@ -2,3 +2,5 @@ streamlit>=1.36.0
 openai>=1.0.0
 python-dotenv>=1.0.0
 pandas>=2.0.0
+scikit-learn>=1.4.0
+matplotlib>=3.8.0