{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# 🎯 [MNT03] 고장률 분석을 통한 다빈도 고장품목 식별\n",
        "### Google Colab 실습 환경 (장비정비·수리부속 / Level 3)\n",
        "\n",
        "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edunlab/milicontent2/blob/main/uploads/MNT03/MNT03_solution.ipynb)\n",
        "\n",
        "---\n",
        "\n",
        "## 📌 1. 프로젝트 개요 및 배경\n",
        "\n",
        "| 항목 | 내용 |\n",
        "| :--- | :--- |\n",
        "| **과정 분류** | 장비정비·수리부속 (Level 3) |\n",
        "| **학습 소요 시간** | 10개 미션 (5일 / 총 28.5시간) |\n",
        "| **환경 사양** | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |\n",
        "\n",
        "### 📖 프로젝트 목표\n",
        "야전 정비 현장에서는 어떤 부품이 자주 고장 나는지 정확히 알아야 수리부속을 먼저 확보하고 예방정비 주기를 조정할 수 있지만, 차량 전자제어장치(ECU)가 남기는 고장 진단 코드(DTC)는 한 번에 여러 개가 겹쳐 기록되고 부품 간 공유 코드도 있어 첫 번째 코드만 보면 원인 부품을 자주 놓친다. 본 프로젝트는 고장·정비 이력을 연결하고 가동시간 대비 고장률을 계산해 다빈도 고장품목을 식별하는 데이터분석 기본기를 익히는 것을 목표로, 교육용 가상 차량 100대의 18개월 DTC·정비 이력을 코드로 만들어 분석한다. 데이터 스키마 정의, 로그 정제, 단일 코드 Baseline과 파레토·상관 분석, 연관 규칙 기반 근본 원인 예측, 정비 조치 추천과 신뢰도 점수, 설명가능성과 정비병용 진단 화면, 숙련 정비관 2차 진단(HITL) 시뮬레이터, 1,000건 동시 주입 부하 시험, AAR 룰 재튜닝과 효과 산출, 재학습을 포함한 통합 도구까지 단계별로 구현한다. 최종 목표는 다빈도 고장품목을 근거와 함께 식별해 수리부속 확보와 예방정비 우선순위를 정하되, 복합 고장의 최종 판단은 숙련 정비관이 내리도록 지원하는 것이다.\n",
        "\n",
        "> **⚠️ [교육용 가상 데이터셋 원칙]**\n",
        "> - 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 **코드 내 난수 기반 가상 데이터로 생성**됩니다.\n",
        "> - 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.\n",
        "\n",
        "---\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# ============================================================================\n",
        "# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트\n",
        "# ============================================================================\n",
        "!pip install -q numpy pandas matplotlib scipy scikit-learn\n",
        "\n",
        "import os\n",
        "import sys\n",
        "import time\n",
        "import math\n",
        "import json\n",
        "import re\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "import warnings\n",
        "warnings.filterwarnings('ignore')\n",
        "\n",
        "# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)\n",
        "plt.rcParams['font.sans-serif'] = 'DejaVu Sans'\n",
        "plt.rcParams['axes.unicode_minus'] = False\n",
        "\n",
        "print(f\"실행 환경 준비 완료 (Python {sys.version.split()[0]})\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 01. 전차/장갑차 전자제어장치(ECU) 고장 진단 코드(DTC) 및 정비 매뉴얼 데이터셋 정의\n",
        "**소요시간**: 약 1.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "전차/장갑차 전자제어장치(ECU) 고장 진단 코드(DTC) 데이터셋을 정의한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **전차/장갑차 계열 가상 차량의 ECU 고장 진단 코드(DTC) 발생 로그, 정비 이력, 정비 매뉴얼 데이터셋의 스키마와 데이터 딕셔너리, 고장률 분석 업무 프로세스 맵을 정의한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **DTC 발생부터 정비 이력 연결, 고장률 집계, 다빈도 품목 식별, 사후검토까지 업무 흐름을 단계별로 정리하고, DTC 로그·정비 건·차량·예측 대상 묶음으로 피처 20개 이상 정의하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `docs/mnt03_logistics_domain_spec.md 및 데이터 딕셔너리` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 군수 실무 업무 프로세스와의 일치성 및 분석 목표의 명확성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 업무 프로세스 맵 작성 완료, 필요 피처(Feature) 목록 20개 이상 정의\n",
        "\n",
        "\n",
        "- 진행 기준: 도메인 분석서 및 데이터 요구명세서 승인\n",
        "\n",
        "\n",
        "- 검증 방법: 군수 담당 실무자 인터뷰 및 기존 전산체계 데이터 구조 대조\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"전차/장갑차 계열 가상 차량의 전자제어장치(ECU) 고장 진단 코드(DTC) 발생 로그, 정비 이력, 정비 매뉴얼 데이터셋의 스키마·데이터 딕셔너리(피처 20개 이상)와 고장률 분석 업무 프로세스 맵을 정의하고, 가상 데이터로 검증한다.\"\"\"\n",
        "import json\n",
        "import re\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# 1) 업무 프로세스 맵: DTC 발생부터 고장률 집계·환류까지 (각 단계에서 생기는 데이터)\n",
        "PROCESS = [(\"DTC 발생\", \"차량 ID, 발생 시각, DTC 코드, 가동시간 계기값\"), (\"로그 수집·정제\", \"중복 제거, 코드·시각 표기 통일\"),\n",
        "           (\"동시 발생 묶음\", \"같은 정비 건에서 발생한 DTC 세트(정규화)\"), (\"고장 진단\", \"추정 고장 부품, 신뢰도\"),\n",
        "           (\"정비 조치\", \"정비 매뉴얼 절차, 수리시간, 교체 부품\"), (\"정비 이력 연결\", \"정비 건 ID로 DTC와 실제 고장 부품 연결\"),\n",
        "           (\"고장률 집계\", \"부품별 1000가동시간당 고장 건수\"), (\"다빈도 품목 식별\", \"파레토(ABC) 등급, 관리 대상 선정\"),\n",
        "           (\"사후검토(AAR)\", \"오진단 사례, 규칙 조정 이력\")]\n",
        "\n",
        "# 2) 스키마(데이터 딕셔너리): 그룹 / 필드 / 자료형 / 규칙 / 설명\n",
        "S = {\n",
        "    \"dtc_log\": {\n",
        "        \"ev_id\": (str, {\"pattern\": r\"^W\\d{5}$\"}, \"정비 건 ID(DTC와 정비 이력을 잇는 키)\"),\n",
        "        \"veh_id\": (str, {\"pattern\": r\"^V\\d{3}$\"}, \"차량 ID(가상)\"),\n",
        "        \"ts\": (float, {\"min\": 0, \"max\": 13000}, \"DTC 발생 시각(모의 시작 후 경과 시간, h)\"),\n",
        "        \"code\": (str, {\"enum\": CODES}, \"DTC 코드(고유·공유·계통 증상 28종)\"),\n",
        "        \"co_count\": (int, {\"min\": 1, \"max\": 8}, \"같은 정비 건에서 동시 발생한 DTC 수\"),\n",
        "    },\n",
        "    \"work_order\": {\n",
        "        \"part_id\": (str, {\"enum\": PID}, \"실제 고장 부품(정비 이력) - 예측대상 1\"),\n",
        "        \"system\": (str, {\"enum\": SYS}, \"고장 계통\"),\n",
        "        \"repair_h\": (float, {\"min\": 0.1, \"max\": 24}, \"수리 소요시간(h)\"),\n",
        "        \"unit_cost\": (float, {\"min\": 0, \"max\": 50}, \"부품 단가지수(가상)\"),\n",
        "        \"unplanned\": (bool, {}, \"비계획·긴급 정비 여부\"),\n",
        "        \"month\": (int, {\"min\": 0, \"max\": 17}, \"정비 월(모의 시작 기준)\"),\n",
        "    },\n",
        "    \"vehicle\": {\n",
        "        \"age\": (float, {\"min\": 0, \"max\": 40}, \"차령(년)\"),\n",
        "        \"op_hours\": (float, {\"min\": 0, \"max\": 400}, \"월 가동시간(h)\"),\n",
        "        \"pm_gap\": (float, {\"min\": 0, \"max\": 500}, \"마지막 예방정비 후 경과 가동시간(h)\"),\n",
        "        \"terrain\": (float, {\"min\": 0, \"max\": 1}, \"운용 지형 가혹도(0~1)\"),\n",
        "        \"load\": (float, {\"min\": 0, \"max\": 1}, \"운용 부하 수준(0~1)\"),\n",
        "        \"temp\": (float, {\"min\": -40, \"max\": 60}, \"정비 시점 기온(도)\"),\n",
        "    },\n",
        "    \"manual\": {\n",
        "        \"code_ref\": (str, {\"enum\": CODES}, \"매뉴얼 기준 DTC 코드\"),\n",
        "        \"cand_parts\": (list, {}, \"코드별 후보 고장 부품 목록\"),\n",
        "        \"step_no\": (int, {\"min\": 1, \"max\": 20}, \"정비 조치 절차 단계 번호\"),\n",
        "        \"safety_flag\": (bool, {}, \"안전 주의 항목 포함 여부\"),\n",
        "    },\n",
        "    \"target\": {\n",
        "        \"fail_rate\": (float, {\"min\": 0, \"max\": 100}, \"부품별 1000가동시간당 고장률 - 예측대상 2\"),\n",
        "        \"confidence\": (float, {\"min\": 0, \"max\": 1}, \"진단 추천 신뢰도 점수 - 예측대상 3\"),\n",
        "    },\n",
        "}\n",
        "\n",
        "\n",
        "def check(name, v, typ, rule):\n",
        "    if not isinstance(v, typ) or (typ in (int, float) and isinstance(v, bool)):\n",
        "        return [f\"{name}: 자료형 오류({type(v).__name__})\"]\n",
        "    e = []\n",
        "    if \"enum\" in rule and v not in rule[\"enum\"]:\n",
        "        e.append(f\"{name}: 허용되지 않은 값 '{v}'\")\n",
        "    if \"min\" in rule and v < rule[\"min\"]:\n",
        "        e.append(f\"{name}: 최소값 {rule['min']} 미만\")\n",
        "    if \"max\" in rule and v > rule[\"max\"]:\n",
        "        e.append(f\"{name}: 최대값 {rule['max']} 초과\")\n",
        "    if \"pattern\" in rule and not re.match(rule[\"pattern\"], v):\n",
        "        e.append(f\"{name}: 형식 불일치 '{v}'\")\n",
        "    return e\n",
        "\n",
        "\n",
        "def validate(rec):\n",
        "    errs = []\n",
        "    for g, fields in S.items():\n",
        "        for k, (typ, rule, _) in fields.items():\n",
        "            if k not in rec.get(g, {}):\n",
        "                errs.append(f\"{g}.{k}: 필수 필드 누락\")\n",
        "            else:\n",
        "                errs += check(f\"{g}.{k}\", rec[g][k], typ, rule)\n",
        "    return errs\n",
        "\n",
        "\n",
        "# 3) [교육용가상자료] 생성 데이터에서 20건 추출해 레코드 형태로 만들고 검증\n",
        "fleet, ev, log = make_data()\n",
        "f = fleet.set_index(\"veh\")\n",
        "recs = []\n",
        "for e in ev.sample(20, random_state=0).itertuples():\n",
        "    first = log[log[\"ev_id\"] == e.ev_id].iloc[0]\n",
        "    cs = e.codes\n",
        "    recs.append({\n",
        "        \"dtc_log\": {\"ev_id\": e.ev_id, \"veh_id\": e.veh, \"ts\": float(round(first[\"ts\"], 1)), \"code\": cs[0], \"co_count\": len(cs)},\n",
        "        \"work_order\": {\"part_id\": e.part, \"system\": PSYS[e.part], \"repair_h\": float(REPH[e.part]), \"unit_cost\": float(COST[e.part]),\n",
        "                       \"unplanned\": bool(len(cs) >= 3), \"month\": int(e.month)},\n",
        "        \"vehicle\": {\"age\": float(f.loc[e.veh, \"age\"]), \"op_hours\": float(round(f.loc[e.veh, \"base_h\"], 1)), \"pm_gap\": float(round(f.loc[e.veh, \"pm_gap\"], 1)),\n",
        "                    \"terrain\": float(round(f.loc[e.veh, \"terrain\"], 2)), \"load\": float(round(f.loc[e.veh, \"load\"], 2)), \"temp\": float(round(e.temp, 1))},\n",
        "        \"manual\": {\"code_ref\": cs[0], \"cand_parts\": [p for p in PID if cs[0] in (SPEC[p], SHARED.get(p))] or [e.part], \"step_no\": 1,\n",
        "                   \"safety_flag\": PSYS[e.part] in CRIT_SYS},\n",
        "        \"target\": {\"fail_rate\": float(RATE[e.part]), \"confidence\": 0.8}})\n",
        "ok = sum(not validate(r) for r in recs)\n",
        "print(f\"가상 데이터 {len(ev)}건(DTC 로그 {len(log)}행) 생성, 표본 {len(recs)}건 중 검증 통과 {ok}건, 오류 {len(recs) - ok}건\")\n",
        "print(f\"전체 DTC 로그 코드 허용값 검증: {'통과' if log['code'].isin(CODES).all() else '실패'}\")\n",
        "\n",
        "# 4) 오류 주입 시험\n",
        "bad = json.loads(json.dumps(recs[0]))\n",
        "bad[\"dtc_log\"][\"code\"] = \"C999\"                  # 정의되지 않은 DTC 코드\n",
        "bad[\"dtc_log\"][\"veh_id\"] = \"TANK-7\"              # ID 형식 오류\n",
        "bad[\"vehicle\"][\"load\"] = 1.8                     # 범위 초과\n",
        "del bad[\"work_order\"][\"repair_h\"]                # 필수 필드 누락\n",
        "bad[\"target\"][\"confidence\"] = \"높음\"             # 자료형 오류\n",
        "print(\"\\n[오류 주입 결과]\")\n",
        "for e in validate(bad):\n",
        "    print(\" -\", e)\n",
        "\n",
        "# 5) 프로세스 맵과 데이터 딕셔너리\n",
        "print(\"\\n[업무 프로세스 맵]\")\n",
        "print(\" -> \".join(f\"{i}.{p}\" for i, (p, _) in enumerate(PROCESS, 1)))\n",
        "for p, d in PROCESS:\n",
        "    print(f\"   {p:10s}: {d}\")\n",
        "rows = [(g, k, t.__name__, d) for g, fl in S.items() for k, (t, r, d) in fl.items()]\n",
        "dd = pd.DataFrame(rows, columns=[\"그룹\", \"필드\", \"자료형\", \"설명\"])\n",
        "print(f\"\\n[데이터 딕셔너리] 피처 {len(dd)}개 (목표 20개 이상: {'달성' if len(dd) >= 20 else '미달'})\")\n",
        "print(dd.to_string(index=False))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "가상 데이터 10681건(DTC 로그 22079행) 생성, 표본 20건 중 검증 통과 20건, 오류 0건\n",
        "전체 DTC 로그 코드 허용값 검증: 통과\n",
        "\n",
        "[오류 주입 결과]\n",
        " - dtc_log.veh_id: 형식 불일치 'TANK-7'\n",
        " - dtc_log.code: 허용되지 않은 값 'C999'\n",
        " - work_order.repair_h: 필수 필드 누락\n",
        " - vehicle.load: 최대값 1 초과\n",
        " - target.confidence: 자료형 오류(str)\n",
        "\n",
        "[업무 프로세스 맵]\n",
        "1.DTC 발생 -> 2.로그 수집·정제 -> 3.동시 발생 묶음 -> 4.고장 진단 -> 5.정비 조치 -> 6.정비 이력 연결 -> 7.고장률 집계 -> 8.다빈도 품목 식별 -> 9.사후검토(AAR)\n",
        "   DTC 발생    : 차량 ID, 발생 시각, DTC 코드, 가동시간 계기값\n",
        "   로그 수집·정제  : 중복 제거, 코드·시각 표기 통일\n",
        "   동시 발생 묶음  : 같은 정비 건에서 발생한 DTC 세트(정규화)\n",
        "   고장 진단     : 추정 고장 부품, 신뢰도\n",
        "   정비 조치     : 정비 매뉴얼 절차, 수리시간, 교체 부품\n",
        "   정비 이력 연결  : 정비 건 ID로 DTC와 실제 고장 부품 연결\n",
        "   고장률 집계    : 부품별 1000가동시간당 고장 건수\n",
        "   다빈도 품목 식별 : 파레토(ABC) 등급, 관리 대상 선정\n",
        "   사후검토(AAR) : 오진단 사례, 규칙 조정 이력\n",
        "\n",
        "[데이터 딕셔너리] 피처 23개 (목표 20개 이상: 달성)\n",
        "        그룹          필드   자료형                          설명\n",
        "   dtc_log       ev_id   str   정비 건 ID(DTC와 정비 이력을 잇는 키)\n",
        "   dtc_log      veh_id   str                   차량 ID(가상)\n",
        "   dtc_log          ts float DTC 발생 시각(모의 시작 후 경과 시간, h)\n",
        "   dtc_log        code   str     DTC 코드(고유·공유·계통 증상 28종)\n",
        "   dtc_log    co_count   int      같은 정비 건에서 동시 발생한 DTC 수\n",
        "work_order     part_id   str    실제 고장 부품(정비 이력) - 예측대상 1\n",
        "work_order      system   str                       고장 계통\n",
        "work_order    repair_h float                  수리 소요시간(h)\n",
        "work_order   unit_cost float                 부품 단가지수(가상)\n",
        "work_order   unplanned  bool                비계획·긴급 정비 여부\n",
        "work_order       month   int              정비 월(모의 시작 기준)\n",
        "   vehicle         age float                       차령(년)\n",
        "   vehicle    op_hours float                   월 가동시간(h)\n",
        "   vehicle      pm_gap float       마지막 예방정비 후 경과 가동시간(h)\n",
        "   vehicle     terrain float              운용 지형 가혹도(0~1)\n",
        "   vehicle        load float               운용 부하 수준(0~1)\n",
        "... (이하 생략)\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 02. DTC 시계열 발생 로그 정제, 동시 발생 코드 정규화 전처리 모듈 구현\n",
        "**소요시간**: 약 2.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "DTC 발생 로그 정제 및 동시 발생 코드 정규화 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **DTC 시계열 발생 로그를 정제하고 같은 정비 건에서 동시 발생한 코드를 정규화하는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **중복 행 제거, 차량 ID·DTC 코드·시각 표기 통일, 가동시간 단위 통일, 음수 가동시간 결측 처리, 미등록 차량·미정의 코드 격리, KNN 결측 보정, 동시 발생 코드의 정비 건 묶음·정렬 등 정제 규칙 10종을 적용 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/etl/mnt03_data_cleansing.py 및 정제 품질 보고서` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 군수 특수 예외(비계획 정비, 긴급 불출 등) 데이터 보존의 타당성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 데이터 정제 규칙 10종 적용, 데이터 정합성 검증 통과율 100%\n",
        "\n",
        "\n",
        "- 진행 기준: ETL 파이프라인 구현 및 정제된 마스터 데이터셋 구축\n",
        "\n",
        "\n",
        "- 검증 방법: 전후 데이터 통계량(기술통계) 비교 및 무결성 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"표기 불일치·중복·단위 혼재·결측·미등록 코드가 섞인 가상 DTC 발생 로그에 정제 규칙 10종을 적용하고, 같은 시간대에 함께 발생한 DTC를 정비 건 단위로 묶어 정규화(중복 제거·정렬)한 마스터 데이터를 만든다. 긴급·비계획 정비는 보존하며 정합성 검증 통과율과 정제 전후 통계를 확인한다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from sklearn.impute import KNNImputer\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# 1) [교육용가상자료] 원천 로그: 생성 로그에 일부러 오류를 섞는다\n",
        "fleet, ev, log = make_data()\n",
        "rng = np.random.default_rng(2)\n",
        "clean_n = len(log)\n",
        "start = pd.Timestamp(\"2025-01-01\")\n",
        "raw = pd.DataFrame({\"veh\": log[\"veh\"].values, \"ts\": [(start + pd.Timedelta(hours=t)).strftime(\"%Y-%m-%d %H:%M\") for t in log[\"ts\"]],\n",
        "                    \"code\": log[\"code\"].values, \"emergency\": False})\n",
        "vn = raw[\"veh\"].str[1:].astype(int).values\n",
        "raw[\"op_h\"] = (vn * 37 % 900) + log[\"ts\"].values * 0.15           # 차량별 누적 가동시간 계기값(가상)\n",
        "raw[\"op_unit\"] = \"h\"\n",
        "raw[\"op_h\"] = raw[\"op_h\"].round(1)\n",
        "idx = rng.choice(len(raw), 2600, replace=False)\n",
        "a, b, c, d_, e, f, g, h_, k = np.split(idx, [400, 900, 1500, 1800, 1880, 1960, 2040, 2300])\n",
        "raw.loc[a, \"veh\"] = rng.choice([\"v\", \" V\"], len(a)) + raw.loc[a, \"veh\"].str[1:]                       # 차량 ID 표기 불일치\n",
        "raw.loc[b, \"code\"] = [x.lower() if i % 2 else x[0] + \"-\" + x[1:] + \" \" for i, x in enumerate(raw.loc[b, \"code\"])]   # 코드 표기 불일치\n",
        "raw.loc[c, \"ts\"] = pd.to_datetime(raw.loc[c, \"ts\"]).dt.strftime(\"%Y/%m/%d %H:%M:%S\")                   # 시각 형식 혼재\n",
        "raw.loc[d_, \"op_h\"] = (raw.loc[d_, \"op_h\"] * 60).round(0); raw.loc[d_, \"op_unit\"] = \"min\"             # 단위 혼재\n",
        "raw.loc[e, \"op_h\"] = -1.0                                                                             # 불가능 값\n",
        "raw.loc[f, \"veh\"] = \"V999\"                                                                            # 미등록 차량\n",
        "raw.loc[g, \"code\"] = \"ZZ9\"                                                                            # 미정의 코드\n",
        "raw.loc[h_, \"op_h\"] = np.nan                                                                          # 결측\n",
        "burst = raw.iloc[k[:40]].copy()                                                                       # 긴급·비계획 정비: DTC 6개 폭주\n",
        "bursts = []\n",
        "for _, r in burst.iterrows():\n",
        "    for j, cd in enumerate(rng.choice(CODES, 6, replace=False)):\n",
        "        bursts.append({**r, \"code\": cd, \"emergency\": True,\n",
        "                       \"ts\": (pd.to_datetime(r[\"ts\"]) + pd.Timedelta(minutes=j)).strftime(\"%Y-%m-%d %H:%M\")})\n",
        "raw = pd.concat([raw, pd.DataFrame(bursts), raw.sample(150, random_state=1)], ignore_index=True)    # 폭주 건 + 중복 150건\n",
        "raw = raw.sample(frac=1, random_state=3).reset_index(drop=True)\n",
        "print(f\"원천 로그 {len(raw)}행 (정상 {clean_n}행 + 긴급 폭주 {len(bursts)}행 + 중복 150행, 오류 주입 {len(idx)}행)\")\n",
        "\n",
        "# 2) 정제 규칙 10종\n",
        "df, applied = raw.copy(), []\n",
        "def rule(name, n):\n",
        "    applied.append((name, int(n)))\n",
        "\n",
        "rule(\"R1 중복 행 제거\", df.duplicated().sum()); df = df.drop_duplicates().reset_index(drop=True)\n",
        "v = df[\"veh\"].str.strip().str.upper(); rule(\"R2 차량 ID 표기 통일\", (v != df[\"veh\"]).sum()); df[\"veh\"] = v\n",
        "cd = df[\"code\"].str.strip().str.upper().str.replace(\"-\", \"\", regex=False); rule(\"R3 DTC 코드 표기 통일\", (cd != df[\"code\"]).sum()); df[\"code\"] = cd\n",
        "t1 = pd.to_datetime(df[\"ts\"], format=\"%Y-%m-%d %H:%M\", errors=\"coerce\")\n",
        "t2 = pd.to_datetime(df[\"ts\"], format=\"%Y/%m/%d %H:%M:%S\", errors=\"coerce\"); rule(\"R4 발생 시각 형식 통일\", t1.isna().sum()); df[\"ts\"] = t1.fillna(t2)\n",
        "mn = df[\"op_unit\"] == \"min\"; df.loc[mn, \"op_h\"] = df.loc[mn, \"op_h\"] / 60; df[\"op_unit\"] = \"h\"; rule(\"R5 가동시간 단위 통일(분 -> 시간)\", mn.sum())\n",
        "bad = df[\"op_h\"] < 0; df.loc[bad, \"op_h\"] = np.nan; rule(\"R6 음수 가동시간 결측 처리\", bad.sum())\n",
        "unk = ~df[\"veh\"].isin(fleet[\"veh\"]) | ~df[\"code\"].isin(CODES); quarantine = df[unk].copy(); df = df[~unk].reset_index(drop=True)\n",
        "rule(\"R7 미등록 차량·미정의 코드 격리(삭제 아님)\", unk.sum())\n",
        "df[\"hrs\"] = (df[\"ts\"] - start).dt.total_seconds() / 3600; df[\"vn\"] = df[\"veh\"].str[1:].astype(int)\n",
        "miss = df[\"op_h\"].isna().sum()\n",
        "df[\"op_h\"] = KNNImputer(n_neighbors=5).fit_transform(df[[\"hrs\", \"vn\", \"op_h\"]].assign(vn=df[\"vn\"] / 10))[:, 2]; rule(\"R8 가동시간 결측 KNN 보정\", miss)\n",
        "df = df.sort_values([\"veh\", \"ts\"]).reset_index(drop=True)\n",
        "df[\"case\"] = (df[\"veh\"] != df[\"veh\"].shift()) | (df[\"ts\"].diff() > pd.Timedelta(hours=1)); df[\"case_id\"] = df[\"case\"].cumsum()\n",
        "before = len(df); df = df.drop_duplicates([\"case_id\", \"code\"]).reset_index(drop=True)\n",
        "cases = df.groupby(\"case_id\").agg(veh=(\"veh\", \"first\"), t=(\"ts\", \"first\"), codes=(\"code\", lambda s: tuple(sorted(s))), n=(\"code\", \"size\"), emergency=(\"emergency\", \"max\"))\n",
        "print(f\"(동시 발생 묶음 과정에서 중복 코드 {before - len(df)}행 제거)\")\n",
        "rule(\"R9 동시 발생 코드 정비 건 묶음·정렬(생성 건수)\", len(cases))\n",
        "cases[\"outlier_flag\"] = (cases[\"n\"] > 4) & ~cases[\"emergency\"]\n",
        "rule(\"R10 긴급·비계획 정비 보존(일반 이상 건은 표시만)\", cases[\"emergency\"].sum())\n",
        "print(\"\\n[적용 규칙과 처리 건수]\")\n",
        "for nme, n in applied:\n",
        "    print(f\" {nme:40s} {n:5d}건\")\n",
        "\n",
        "# 3) 정합성 검증\n",
        "checks = {\"중복 0건\": not df.duplicated([\"case_id\", \"code\"]).any(), \"차량 ID 형식\": df[\"veh\"].str.match(r\"^V\\d{3}$\").all(),\n",
        "          \"코드 허용 범위\": df[\"code\"].isin(CODES).all(), \"시각 변환\": df[\"ts\"].notna().all(), \"가동시간 단위 h\": (df[\"op_unit\"] == \"h\").all(),\n",
        "          \"가동시간 0 이상\": (df[\"op_h\"] >= 0).all(), \"미등록 건 격리 보관\": len(quarantine) == unk.sum(),\n",
        "          \"결측 0건\": df[[\"veh\", \"code\", \"ts\", \"op_h\"]].notna().all().all(), \"묶음 내 코드 정렬\": all(list(c) == sorted(c) for c in cases[\"codes\"]),\n",
        "          f\"긴급 정비 {len(bursts) // 6}건 보존\": int(cases[\"emergency\"].sum()) >= len(bursts) // 6}\n",
        "rate = np.mean(list(checks.values()))\n",
        "print(f\"\\n정합성 검증 {sum(checks.values())}/{len(checks)} 통과, 통과율 {rate:.0%} (목표 100%: {'달성' if rate == 1 else '미달'})\")\n",
        "for kx, vx in checks.items():\n",
        "    if not vx:\n",
        "        print(\" 실패:\", kx)\n",
        "print(f\"정비 건 {len(cases)}건 생성(평균 DTC {cases['n'].mean():.2f}개), 이상 표시 {int(cases['outlier_flag'].sum())}건\")\n",
        "\n",
        "# 4) 정제 전후 기술통계 비교\n",
        "num = pd.to_numeric(raw[\"op_h\"], errors=\"coerce\")\n",
        "cmp = pd.DataFrame({\"정제 전\": [len(raw), raw[\"veh\"].nunique(), raw[\"code\"].nunique(), num.mean(), num.min(), num.isna().sum()],\n",
        "                    \"정제 후\": [len(df), df[\"veh\"].nunique(), df[\"code\"].nunique(), df[\"op_h\"].mean(), df[\"op_h\"].min(), df[\"op_h\"].isna().sum()]},\n",
        "                   index=[\"행 수\", \"차량 종류 수\", \"코드 종류 수\", \"가동시간 평균(h)\", \"가동시간 최소(h)\", \"결측 수\"]).round(1)\n",
        "print(\"\\n[정제 전후 기술통계]\"); print(cmp.to_string())\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 3.5))\n",
        "ax[0].hist(num.dropna(), bins=40, alpha=0.6, label=\"raw\"); ax[0].hist(df[\"op_h\"], bins=40, alpha=0.6, label=\"cleaned\")\n",
        "ax[0].set_title(\"Operating hours before/after unit fix\"); ax[0].legend()\n",
        "ax[1].bar([n.split()[0] for n, _ in applied], [n for _, n in applied]); ax[1].set_title(\"Records changed by rule\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "원천 로그 22469행 (정상 22079행 + 긴급 폭주 240행 + 중복 150행, 오류 주입 2600행)\n",
        "(동시 발생 묶음 과정에서 중복 코드 68행 제거)\n",
        "\n",
        "[적용 규칙과 처리 건수]\n",
        " R1 중복 행 제거                                 150건\n",
        " R2 차량 ID 표기 통일                             400건\n",
        " R3 DTC 코드 표기 통일                            500건\n",
        " R4 발생 시각 형식 통일                             600건\n",
        " R5 가동시간 단위 통일(분 -> 시간)                     300건\n",
        " R6 음수 가동시간 결측 처리                            80건\n",
        " R7 미등록 차량·미정의 코드 격리(삭제 아님)                 160건\n",
        " R8 가동시간 결측 KNN 보정                          340건\n",
        " R9 동시 발생 코드 정비 건 묶음·정렬(생성 건수)            10504건\n",
        " R10 긴급·비계획 정비 보존(일반 이상 건은 표시만)              40건\n",
        "\n",
        "정합성 검증 10/10 통과, 통과율 100% (목표 100%: 달성)\n",
        "정비 건 10504건 생성(평균 DTC 2.10개), 이상 표시 85건\n",
        "\n",
        "[정제 전후 기술통계]\n",
        "               정제 전     정제 후\n",
        "행 수         22469.0  22091.0\n",
        "차량 종류 수       252.0    100.0\n",
        "코드 종류 수        82.0     28.0\n",
        "가동시간 평균(h)   2496.2   1372.4\n",
        "가동시간 최소(h)     -1.0      3.8\n",
        "결측 수          261.0      0.0\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 03. 단일 고장 코드 기반 1차 고장 부품 매핑 Baseline 알고리즘 구현\n",
        "**소요시간**: 약 2.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "단일 고장 코드 기반 1차 고장 부품 매핑 Baseline 알고리즘을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **단일 고장 코드 기반 1차 고장 부품 매핑 Baseline(첫 기록 코드와 가장 자주 함께 나온 부품)을 구현하고, 탐색적 분석으로 다빈도 고장품목과 고장 증가 요인을 찾는다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **학습 기간 이력으로 코드와 부품의 대응표를 만들어 검증 기간의 부품 정확도를 측정하고, 공유·증상 코드처럼 모호한 코드가 첫 기록일 때 정확도가 떨어지는지 확인하고 유효성을 점검하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `notebooks/mnt03_eda_analysis.ipynb 및 EDA 요약 리포트` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 군수 의사결정에 직관적인 인사이트 제공 수준\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: EDA 시각화 차트 8종 이상 생성, 유의미한 상관관계 피처 5개 이상 도출\n",
        "\n",
        "\n",
        "- 진행 기준: EDA 노트북 실행 완료 및 핵심 발견점(Key Findings) 문서화\n",
        "\n",
        "\n",
        "- 검증 방법: 통계적 가설 검정(p-value, 상관계수)을 통한 유의성 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"단일 DTC 코드만 보고 가장 흔한 고장 부품을 고르는 1차 Baseline 매핑 알고리즘을 구현해 정확도와 한계를 확인하고, 탐색적 분석(EDA 차트 8종, 고장 건수와 피처의 상관계수·p-value, 파레토 ABC 분류)으로 1000가동시간당 고장률이 높은 다빈도 고장품목을 식별한다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy.stats import pearsonr\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "ev[\"first\"] = ev[\"codes\"].str[0]\n",
        "train, test = ev[ev[\"month\"] < 12], ev[ev[\"month\"] >= 12]\n",
        "\n",
        "# 1) Baseline: 첫 번째로 기록된 코드 -> 학습 기간에 그 코드와 가장 자주 함께 나온 부품\n",
        "mp = train.groupby(\"first\")[\"part\"].agg(lambda s: s.value_counts().index[0])\n",
        "pred = test[\"first\"].map(mp)\n",
        "acc = (pred == test[\"part\"]).mean()\n",
        "sys_acc = (pred.map(PSYS) == test[\"part\"].map(PSYS)).mean()\n",
        "print(f\"[교육용가상자료] 정비 건 {len(ev)}건, DTC 로그 {len(log)}행, 차량 {len(fleet)}대, 18개월\")\n",
        "print(f\"Baseline 부품 정확도 {acc:.1%}, 고장 계통 정확도 {sys_acc:.1%} (검증 {len(test)}건)\")\n",
        "amb = test[\"first\"].isin(list(SHARED.values()) + list(SYMP.values()))\n",
        "print(f\"  모호한 코드(공유·증상) 첫 기록 건 정확도 {(pred[amb] == test['part'][amb]).mean():.1%} vs 고유 코드 {(pred[~amb] == test['part'][~amb]).mean():.1%}\")\n",
        "\n",
        "# 2) 부품별 고장률(1000가동시간당)과 파레토(ABC)\n",
        "tot_h = fleet[\"hours\"].sum()\n",
        "cnt = ev[\"part\"].value_counts().reindex(PID)\n",
        "rate = (cnt / tot_h * 1000).round(3)\n",
        "par = pd.DataFrame({\"부품\": [PNAME[p] for p in PID], \"건수\": cnt.values, \"고장률\": rate.values}, index=PID).sort_values(\"건수\", ascending=False)\n",
        "par[\"누적비율\"] = par[\"건수\"].cumsum() / par[\"건수\"].sum()\n",
        "par[\"등급\"] = np.where(par[\"누적비율\"] <= 0.80, \"A\", np.where(par[\"누적비율\"] <= 0.95, \"B\", \"C\"))\n",
        "par.loc[par.index[0], \"등급\"] = \"A\"\n",
        "print(f\"\\n[파레토] 총 가동 {tot_h:,.0f}시간, 전체 고장률 {cnt.sum() / tot_h * 1000:.1f}건/1000h\")\n",
        "print(par.head(7).assign(누적비율=lambda d: d[\"누적비율\"].map(\"{:.0%}\".format)).to_string())\n",
        "top20 = par.head(int(np.ceil(len(par) * 0.2)))\n",
        "print(f\"상위 20% 품목({len(top20)}종: {', '.join(top20['부품'])})이 전체 고장의 {top20['건수'].sum() / cnt.sum():.0%}, A등급 {int((par['등급'] == 'A').sum())}종\")\n",
        "\n",
        "# 3) 상관분석: 차량별 정비 건수와 피처\n",
        "vc = ev.groupby(\"veh\").size().reindex(fleet[\"veh\"]).fillna(0).values\n",
        "rows = []\n",
        "for col, lab in [(\"age\", \"차령\"), (\"pm_gap\", \"예방정비 경과시간\"), (\"terrain\", \"지형 가혹도\"), (\"load\", \"운용 부하\"), (\"hours\", \"총 가동시간\"), (\"humid\", \"습도 지수\")]:\n",
        "    r, p = pearsonr(fleet[col], vc); rows.append((lab, r, p))\n",
        "mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "tm = ev.groupby(\"month\")[\"temp\"].mean()\n",
        "r, p = pearsonr(tm, mc[\"P03\"]); rows.append((\"월평균 기온-축전지 고장\", r, p))\n",
        "cor = pd.DataFrame(rows, columns=[\"피처\", \"r\", \"p\"])\n",
        "cor[\"유의\"] = cor[\"p\"] < 0.05\n",
        "print(\"\\n[상관분석] 피처 | r | p-value | 유의(p<0.05)\")\n",
        "for x in cor.itertuples():\n",
        "    print(f\" {x.피처:14s} {x.r:+.2f}  {x.p:.4f}  {'유의' if x.유의 else '-'}\")\n",
        "nsig = int(cor[\"유의\"].sum())\n",
        "\n",
        "# 4) EDA 차트 8종\n",
        "fig, ax = plt.subplots(2, 4, figsize=(16, 7))\n",
        "ax[0, 0].bar(par.index, par[\"고장률\"]); ax[0, 0].set_title(\"1 Failure rate per 1000h\")\n",
        "ax[0, 1].plot(range(len(par)), par[\"누적비율\"], marker=\"o\"); ax[0, 1].axhline(0.8, ls=\"--\", color=\"k\"); ax[0, 1].set_title(\"2 Pareto cumulative share\")\n",
        "mc.sum(axis=1).plot(ax=ax[0, 2]); ax[0, 2].set_title(\"3 Monthly failures\")\n",
        "ev[\"codes\"].str.len().value_counts().sort_index().plot.bar(ax=ax[0, 3]); ax[0, 3].set_title(\"4 DTCs per maintenance case\")\n",
        "log[\"code\"].value_counts().head(10).plot.bar(ax=ax[1, 0]); ax[1, 0].set_title(\"5 Top DTC codes\")\n",
        "ev[\"part\"].map(PSYS).value_counts().plot.bar(ax=ax[1, 1]); ax[1, 1].set_title(\"6 Failures by system\")\n",
        "ax[1, 2].scatter(fleet[\"age\"], vc, s=8); ax[1, 2].set_title(\"7 Vehicle age vs failures\")\n",
        "ax[1, 3].scatter(fleet[\"pm_gap\"], vc, s=8); ax[1, 3].set_title(\"8 PM gap vs failures\")\n",
        "plt.tight_layout(); plt.show()\n",
        "\n",
        "print(f\"\\nEDA 차트 8종 생성(목표 8종 이상: 달성), 유의한 피처 {nsig}개 (목표 5개 이상: {'달성' if nsig >= 5 else '미달'})\")\n",
        "print(\"핵심 발견: 1) 소수 부품에 고장 집중 2) 공유·증상 코드가 첫 기록이면 Baseline이 자주 틀림 3) 고령·가혹 지형·예방정비 지연 차량의 고장 증가\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 정비 건 10681건, DTC 로그 22079행, 차량 100대, 18개월\n",
        "Baseline 부품 정확도 68.7%, 고장 계통 정확도 96.8% (검증 3890건)\n",
        "  모호한 코드(공유·증상) 첫 기록 건 정확도 59.7% vs 고유 코드 80.6%\n",
        "\n",
        "[파레토] 총 가동 171,649시간, 전체 고장률 62.2건/1000h\n",
        "           부품    건수     고장률 누적비율 등급\n",
        "P01     연료분사기  2800  16.312  26%  A\n",
        "P02      터보차저  1985  11.564  45%  A\n",
        "P03       축전지  1983  11.553  63%  A\n",
        "P04       발전기  1244   7.247  75%  A\n",
        "P05  변속기 제어모듈   665   3.874  81%  B\n",
        "P06    냉각수 펌프   477   2.779  86%  B\n",
        "P07     유압 펌프   390   2.272  89%  B\n",
        "상위 20% 품목(4종: 연료분사기, 터보차저, 축전지, 발전기)이 전체 고장의 75%, A등급 4종\n",
        "\n",
        "[상관분석] 피처 | r | p-value | 유의(p<0.05)\n",
        " 차령             +0.41  0.0000  유의\n",
        " 예방정비 경과시간      +0.57  0.0000  유의\n",
        " 지형 가혹도         +0.40  0.0000  유의\n",
        " 운용 부하          +0.35  0.0004  유의\n",
        " 총 가동시간         +0.32  0.0011  유의\n",
        " 습도 지수          +0.07  0.5174  -\n",
        " 월평균 기온-축전지 고장  -0.95  0.0000  유의\n",
        "\n",
        "EDA 차트 8종 생성(목표 8종 이상: 달성), 유의한 피처 6개 (목표 5개 이상: 달성)\n",
        "핵심 발견: 1) 소수 부품에 고장 집중 2) 공유·증상 코드가 첫 기록이면 Baseline이 자주 틀림 3) 고령·가혹 지형·예방정비 지연 차량의 고장 증가\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 04. 연관 규칙 마이닝(Apriori/FP-Growth) 기반 다중 DTC 동시 발생 시 근본 원인 예측 모델 고도화\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "연관 규칙 마이닝 기반 다중 DTC 동시 발생 시 근본 원인 예측 모델을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **연관 규칙 마이닝(Apriori)으로 다중 DTC 동시 발생 시 근본 원인 부품을 예측하는 모델을 고도화하고, 월간 고장 건수 기준 모델의 파생 피처를 설계한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **동시 발생 DTC 세트와 근본 원인 부품을 항목으로 보고 최소 지지도 미만을 가지치기하며 빈발 항목집합과 연관 규칙(신뢰도·향상도)을 구 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/features/mnt03_feature_engineering.py 및 기준 모델` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 군수 실무에서 납득 가능한 피처 설계의 논리성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 파생 피처 10개 이상 생성, 기준 모델 MAPE 25% 이하 달성\n",
        "\n",
        "\n",
        "- 진행 기준: 피처 파이프라인 및 베이스라인 모델 성능 지표 도출\n",
        "\n",
        "\n",
        "- 검증 방법: 교차 검증(Time-series Cross Validation)을 통한 과적합 방지\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"Apriori 연관 규칙 마이닝으로 여러 DTC가 동시에 나타났을 때의 근본 원인 부품을 예측하는 모델을 만들어 단일 코드 Baseline과 비교하고, 부품별 월간 고장 건수 예측용 파생 피처 10개 이상으로 기준 모델을 학습해 시계열 교차검증 MAPE와 피처 중요도를 확인한다.\"\"\"\n",
        "import itertools\n",
        "from collections import Counter\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from sklearn.linear_model import Ridge\n",
        "from sklearn.preprocessing import StandardScaler\n",
        "from sklearn.pipeline import make_pipeline\n",
        "from sklearn.inspection import permutation_importance\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "ev[\"first\"] = ev[\"codes\"].str[0]\n",
        "train, test = ev[ev[\"month\"] < 12], ev[ev[\"month\"] >= 12]\n",
        "\n",
        "\n",
        "# 1) Apriori: 항목 = DTC 코드 + 근본 원인 항목(ROOT:부품). 수준별로 후보를 만들고 최소 지지도 미만은 가지치기한다.\n",
        "def apriori(trans, min_sup, max_len=3):\n",
        "    n = len(trans)\n",
        "    cnt = Counter(i for t in trans for i in t)\n",
        "    freq = {frozenset([i]): c / n for i, c in cnt.items() if c / n >= min_sup}\n",
        "    allf, cur = dict(freq), list(freq)\n",
        "    for k in range(2, max_len + 1):\n",
        "        cands = {a | b for a, b in itertools.combinations(cur, 2) if len(a | b) == k}\n",
        "        cands = {c for c in cands if all(frozenset(s) in allf for s in itertools.combinations(c, k - 1))}   # 부분집합 가지치기\n",
        "        cc = Counter(c for t in trans for c in cands if c <= t)\n",
        "        cur = [c for c, v in cc.items() if v / n >= min_sup]\n",
        "        allf.update({c: cc[c] / n for c in cur})\n",
        "    return allf\n",
        "\n",
        "\n",
        "trans = [frozenset(cs) | {\"ROOT:\" + p} for cs, p in zip(train[\"codes\"], train[\"part\"])]\n",
        "F = apriori(trans, min_sup=0.004)\n",
        "rules = []\n",
        "for s, sup in F.items():\n",
        "    roots = [i for i in s if i.startswith(\"ROOT:\")]\n",
        "    if len(roots) == 1 and len(s) >= 2:\n",
        "        ante = s - set(roots)\n",
        "        conf = sup / F[ante]\n",
        "        lift = conf / F[frozenset(roots)]\n",
        "        rules.append({\"ante\": ante, \"part\": roots[0][5:], \"sup\": sup, \"conf\": conf, \"lift\": lift})\n",
        "R = pd.DataFrame(rules)\n",
        "R = R[(R[\"conf\"] >= 0.5) & (R[\"lift\"] > 1.5)].sort_values(\"conf\", ascending=False).reset_index(drop=True)\n",
        "print(f\"[교육용가상자료] 학습 정비 건 {len(train)}건: 빈발 항목집합 {len(F)}개 -> 연관 규칙 {len(R)}개 (신뢰도 0.5 이상, 향상도 1.5 초과)\")\n",
        "for r in R.head(4).itertuples():\n",
        "    print(f\"  {{{', '.join(sorted(r.ante))}}} -> {PNAME[r.part]}  지지도 {r.sup:.3f} 신뢰도 {r.conf:.2f} 향상도 {r.lift:.1f}\")\n",
        "\n",
        "mp = train.groupby(\"first\")[\"part\"].agg(lambda s: s.value_counts().index[0])\n",
        "rl = sorted(zip(R[\"ante\"], R[\"part\"], R[\"conf\"]), key=lambda x: (-x[2], -len(x[0])))\n",
        "\n",
        "\n",
        "def rule_predict(codes, fallback):\n",
        "    for a, p, c in rl:\n",
        "        if a <= set(codes):\n",
        "            return p, c\n",
        "    return fallback, 0.0\n",
        "\n",
        "\n",
        "pr = [rule_predict(cs, mp.get(f, \"P01\")) for cs, f in zip(test[\"codes\"], test[\"first\"])]\n",
        "pred_rule = np.array([p for p, _ in pr]); pred_base = test[\"first\"].map(mp).values\n",
        "a_b, a_r = (pred_base == test[\"part\"].values).mean(), (pred_rule == test[\"part\"].values).mean()\n",
        "multi = test[\"codes\"].str.len().values >= 2\n",
        "print(f\"\\n근본 원인 정확도: 단일 코드 Baseline {a_b:.1%} -> 연관 규칙 {a_r:.1%} (코드 2개 이상 동시 발생 건 {(pred_base[multi] == test['part'].values[multi]).mean():.1%} -> {(pred_rule[multi] == test['part'].values[multi]).mean():.1%})\")\n",
        "\n",
        "# 2) 파생 피처 + 월간 고장 건수 기준 모델(상위 8개 부품, 건수는 로그 변환해 학습)\n",
        "D, FE, top = build_monthly(ev, fleet)\n",
        "print(f\"\\n월간 고장 건수 예측: 파생 피처 {len(FE)}개 (목표 10개 이상: {'달성' if len(FE) >= 10 else '미달'}), 대상 {len(top)}개 부품 x {MONTHS - 3}개월\")\n",
        "\n",
        "\n",
        "def model():\n",
        "    return make_pipeline(StandardScaler(), Ridge(alpha=3.0))\n",
        "\n",
        "\n",
        "def mape(a, p):\n",
        "    return float(np.mean(np.abs(a - p) / a))\n",
        "\n",
        "\n",
        "# 3) 시계열 교차검증: 확장 윈도(과거로만 학습, 뒤 2개월 검증)\n",
        "folds, mapes = [(range(3, s), range(s, s + 2)) for s in (11, 13, 15)], []\n",
        "for tr, te in folds:\n",
        "    a, b = D[D[\"month\"].isin(tr)], D[D[\"month\"].isin(te)]\n",
        "    mapes.append(mape(b[\"y\"].values, np.exp(model().fit(a[FE], np.log(a[\"y\"])).predict(b[FE]))))\n",
        "cv = float(np.mean(mapes))\n",
        "print(\"시계열 교차검증 폴드별 MAPE: \" + \", \".join(f\"{m:.1%}\" for m in mapes) + f\" -> 평균 {cv:.1%} (목표 25% 이하: {'달성' if cv <= 0.25 else '미달'})\")\n",
        "\n",
        "# 4) 피처 중요도(순열 중요도)로 차원 축소\n",
        "a, b = D[D[\"month\"] < 13], D[D[\"month\"] >= 13]\n",
        "m = model().fit(a[FE], np.log(a[\"y\"]))\n",
        "imp = permutation_importance(m, b[FE], np.log(b[\"y\"]), n_repeats=20, random_state=0, scoring=\"neg_mean_absolute_error\").importances_mean\n",
        "order = np.argsort(-imp)\n",
        "keep = [FE[i] for i in order[:8]]\n",
        "m5 = model().fit(a[keep], np.log(a[\"y\"]))\n",
        "full, red = mape(b[\"y\"].values, np.exp(m.predict(b[FE]))), mape(b[\"y\"].values, np.exp(m5.predict(b[keep])))\n",
        "print(\"피처 중요도 상위 5: \" + \", \".join(f\"{FE[i]}({imp[i]:.2f})\" for i in order[:5]))\n",
        "print(f\"차원 축소 {len(FE)}개 -> 8개: 검증 MAPE {full:.1%} -> {red:.1%}\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 3.5))\n",
        "ax[0].bar([\"baseline\", \"rules\"], [a_b, a_r]); ax[0].set_title(\"Root-cause accuracy\")\n",
        "ax[1].barh([FE[i] for i in order[:8]][::-1], imp[order[:8]][::-1]); ax[1].set_title(\"Permutation importance (log MAE)\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 학습 정비 건 6791건: 빈발 항목집합 188개 -> 연관 규칙 57개 (신뢰도 0.5 이상, 향상도 1.5 초과)\n",
        "  {G04} -> 유압 펌프  지지도 0.020 신뢰도 1.00 향상도 29.1\n",
        "  {G05} -> 제동 밸브  지지도 0.007 신뢰도 1.00 향상도 87.1\n",
        "  {C051, G03} -> 변속기 제어모듈  지지도 0.021 신뢰도 1.00 향상도 15.5\n",
        "  {C071, G04} -> 유압 펌프  지지도 0.013 신뢰도 1.00 향상도 29.1\n",
        "\n",
        "근본 원인 정확도: 단일 코드 Baseline 68.7% -> 연관 규칙 78.3% (코드 2개 이상 동시 발생 건 67.1% -> 81.1%)\n",
        "\n",
        "월간 고장 건수 예측: 파생 피처 13개 (목표 10개 이상: 달성), 대상 8개 부품 x 15개월\n",
        "시계열 교차검증 폴드별 MAPE: 28.9%, 17.3%, 17.8% -> 평균 21.3% (목표 25% 이하: 달성)\n",
        "피처 중요도 상위 5: lag1(0.21), log_expected(0.18), lag2(0.14), cold_elec(0.09), rate_lag1(0.07)\n",
        "차원 축소 13개 -> 8개: 검증 MAPE 18.6% -> 21.6%\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 05. DTC 기반 정비 조치 가이드라인 및 추천 신뢰도 점수 산출 엔진 개발\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "DTC 기반 정비 조치 가이드라인 및 추천 신뢰도 점수 산출 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **DTC 입력으로 정비 조치 가이드라인과 추천 신뢰도 점수를 산출하는 엔진을 개발하고 기준 모델 대비 개선을 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **코드 유형별 가중치를 둔 고장 부품 확률 모델로 상위 3개 후보와 신뢰도를 내고, 후보별 정비 조치 절차를 진단 카드로 출력 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/models/mnt03_advanced_model.py` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 비정형/불규칙 수요에 대한 모델의 적응력 및 예측 안정성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 기준 모델 대비 오차율(MAE/RMSE) 20% 이상 개선, R2 Score 0.85 이상\n",
        "\n",
        "\n",
        "- 진행 기준: 모델 하이퍼파라미터 튜닝 완료 및 최적 모델 가중치 저장\n",
        "\n",
        "\n",
        "- 검증 방법: Hold-out 테스트 데이터셋 대상 최종 성능 평가\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"DTC 조합과 차량 맥락을 입력받아 후보 고장 부품 3개와 추천 신뢰도 점수, 정비 조치 가이드라인을 산출하는 추천 엔진을 개발하고, 신뢰도가 실제 정확도와 맞는지(보정) 확인한다. 부품별 월간 고장 건수 예측은 전월값 기준 모델 대비 릿지·그래디언트 부스팅 앙상블의 MAE/RMSE 개선율과 R2로 평가한다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from sklearn.linear_model import LogisticRegression, Ridge\n",
        "from sklearn.ensemble import GradientBoostingRegressor\n",
        "from sklearn.preprocessing import StandardScaler\n",
        "from sklearn.pipeline import make_pipeline\n",
        "from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 부품별 정비 조치 가이드라인(요약 3단계). 실제 정비 교범이 아니라 교육용으로 만든 예시 문구다.\n",
        "# ---------------------------------------------------------------------------\n",
        "GUIDE = {\n",
        "    \"P01\": \"연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체\", \"P02\": \"흡기 압력·이물 점검 -> 축 유격 확인 -> 불량 시 터보차저 교체\",\n",
        "    \"P03\": \"단자 전압·부식 점검 -> 충전 계통 확인 -> 불량 시 축전지 교체\", \"P04\": \"발전 전압 측정 -> 벨트·배선 점검 -> 불량 시 발전기 교체\",\n",
        "    \"P05\": \"제어모듈 소프트웨어·커넥터 점검 -> 변속 이력 확인 -> 불량 시 모듈 교체\", \"P06\": \"냉각수량·누수 점검 -> 펌프 작동음 확인 -> 불량 시 펌프 교체\",\n",
        "    \"P07\": \"유압 오일량·압력 점검 -> 펌프 누설 확인 -> 불량 시 펌프 교체\", \"P08\": \"센서 배선·간극 점검 -> 신호 파형 확인 -> 불량 시 센서 교체\",\n",
        "    \"P09\": \"커넥터 접촉·부식 점검 -> 핀 상태 확인 -> 불량 시 커넥터 교체\", \"P10\": \"필터 오염도 점검 -> 연료 흐름 확인 -> 불량 시 필터 교체\",\n",
        "    \"P11\": \"제동 압력 점검 -> 밸브 작동 확인 -> 불량 시 밸브 교체(안전 확인 필수)\", \"P12\": \"조향 반응 점검 -> 액추에이터 작동 확인 -> 불량 시 교체(안전 확인 필수)\",\n",
        "    \"P13\": \"필터 차압 점검 -> 오염 확인 -> 불량 시 필터 교체\", \"P14\": \"통신 링크 상태 점검 -> 모듈 전원 확인 -> 불량 시 모듈 교체\",\n",
        "    \"P15\": \"압력 센서 신호 점검 -> 윤활유량 확인 -> 불량 시 센서 교체\", \"P16\": \"팬 모터 전류 점검 -> 회전 확인 -> 불량 시 모터 교체\"}\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "train, test = ev[ev[\"month\"] < SPLIT_M], ev[ev[\"month\"] >= SPLIT_M]\n",
        "\n",
        "# 1) 추천 엔진: DTC 멀티핫 + 맥락 -> 부품별 확률(= 추천 신뢰도 점수)\n",
        "Xtr, Xte = feats(train, fleet), feats(test, fleet)\n",
        "clf = LogisticRegression(C=3, max_iter=500).fit(Xtr, train[\"part\"])\n",
        "P = clf.predict_proba(Xte)\n",
        "cls = clf.classes_\n",
        "top1 = cls[P.argmax(axis=1)]\n",
        "conf = P.max(axis=1)\n",
        "y = test[\"part\"].values\n",
        "top3 = np.mean([t in cls[np.argsort(-p)[:3]] for t, p in zip(y, P)])\n",
        "mp = train.assign(first=train[\"codes\"].str[0]).groupby(\"first\")[\"part\"].agg(lambda s: s.value_counts().index[0])\n",
        "base_acc = (test[\"codes\"].str[0].map(mp).values == y).mean()\n",
        "print(f\"[교육용가상자료] 추천 엔진 검증 {len(test)}건: top-1 정확도 {(top1 == y).mean():.1%} (단일 코드 Baseline {base_acc:.1%}), top-3 포함률 {top3:.1%}\")\n",
        "\n",
        "# 2) 신뢰도 점수 보정: 신뢰도 구간별 실제 정확도\n",
        "bins = [0, 0.5, 0.7, 0.9, 1.01]\n",
        "lab = [\"~0.5\", \"0.5~0.7\", \"0.7~0.9\", \"0.9~\"]\n",
        "cb = pd.cut(conf, bins, labels=lab, right=False)\n",
        "cal = pd.DataFrame({\"conf\": conf, \"ok\": top1 == y, \"b\": cb}).groupby(\"b\", observed=True).agg(건수=(\"ok\", \"size\"), 평균신뢰도=(\"conf\", \"mean\"), 실제정확도=(\"ok\", \"mean\"))\n",
        "print(\"\\n[신뢰도 보정] 구간 | 건수 | 평균 신뢰도 | 실제 정확도\")\n",
        "for k, r in cal.iterrows():\n",
        "    print(f\" {k:8s} {int(r['건수']):5d}  {r['평균신뢰도']:.2f}  {r['실제정확도']:.2f}\")\n",
        "ece = float((cal[\"건수\"] * (cal[\"평균신뢰도\"] - cal[\"실제정확도\"]).abs()).sum() / cal[\"건수\"].sum())\n",
        "print(f\"기대 보정 오차(ECE) {ece:.3f}\")\n",
        "\n",
        "# 3) 정비 조치 가이드라인 카드(상위 후보 3개와 신뢰도)\n",
        "print(\"\\n[진단 카드 예시] 정비 조치 가이드라인\")\n",
        "for i in (int(np.argmin(np.abs(conf - 0.95))), int(np.argmin(np.abs(conf - 0.55)))):\n",
        "    order = np.argsort(-P[i])[:3]\n",
        "    print(f\" 입력 DTC {test['codes'].iloc[i]} -> \" + \", \".join(f\"{PNAME[cls[j]]} {P[i, j]:.0%}\" for j in order))\n",
        "    print(f\"   조치: {GUIDE[cls[order[0]]]}\" + (\"\" if conf[i] >= 0.7 else \"  (신뢰도 낮음: 숙련 정비관 확인 권고)\"))\n",
        "\n",
        "# 4) 월간 고장 건수 예측: 기준(전월값) vs 고도화(릿지 + GBR 앙상블, 소규모 격자 탐색)\n",
        "D, FE, top = build_monthly(ev, fleet)\n",
        "a, b = D[D[\"month\"] < 13], D[D[\"month\"] >= 13]\n",
        "ya, yb = np.log(a[\"y\"]), b[\"y\"].values\n",
        "naive = np.expm1(b[\"lag1\"].values)\n",
        "def ens(tr, te, al, dep):\n",
        "    g = GradientBoostingRegressor(n_estimators=150, max_depth=dep, learning_rate=0.05, random_state=0).fit(tr[FE], np.log(tr[\"y\"]))\n",
        "    r = make_pipeline(StandardScaler(), Ridge(alpha=al)).fit(tr[FE], np.log(tr[\"y\"]))\n",
        "    return np.exp(0.5 * g.predict(te[FE]) + 0.5 * r.predict(te[FE]))\n",
        "\n",
        "\n",
        "inner_tr, inner_va = a[a[\"month\"] < 11], a[a[\"month\"] >= 11]          # 하이퍼파라미터는 학습 구간 안에서만 선택(검증 구간 미사용)\n",
        "score = {(al, dep): mean_absolute_error(inner_va[\"y\"], ens(inner_tr, inner_va, al, dep)) for al in (1, 3, 10) for dep in (2, 3)}\n",
        "al, dep = min(score, key=score.get)\n",
        "adv = ens(a, b, al, dep)\n",
        "met = lambda p: (mean_absolute_error(yb, p), mean_squared_error(yb, p) ** 0.5, r2_score(yb, p))\n",
        "m0, m1 = met(naive), met(adv)\n",
        "print(f\"\\n[월간 고장 건수 예측] 검증 {len(b)}건(상위 {len(top)}개 부품 x 5개월), 선택 하이퍼파라미터 ridge alpha={al}, GBR depth={dep}\")\n",
        "print(f\" 기준(전월값)  MAE {m0[0]:.2f}  RMSE {m0[1]:.2f}  R2 {m0[2]:.3f}\")\n",
        "print(f\" 고도화(앙상블) MAE {m1[0]:.2f}  RMSE {m1[1]:.2f}  R2 {m1[2]:.3f}\")\n",
        "imp_mae, imp_rmse = 1 - m1[0] / m0[0], 1 - m1[1] / m0[1]\n",
        "print(f\"개선율 MAE {imp_mae:.1%}, RMSE {imp_rmse:.1%} (목표 20% 이상: {'달성' if min(imp_mae, imp_rmse) >= 0.2 else '미달'}), R2 {m1[2]:.3f} (목표 0.85 이상: {'달성' if m1[2] >= 0.85 else '미달'})\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 3.5))\n",
        "ax[0].plot([0, 1], [0, 1], \"k--\"); ax[0].plot(cal[\"평균신뢰도\"], cal[\"실제정확도\"], marker=\"o\"); ax[0].set_title(\"Confidence calibration\")\n",
        "ax[1].scatter(yb, adv, s=12, label=\"ensemble\"); ax[1].scatter(yb, naive, s=12, label=\"naive\"); ax[1].plot([0, yb.max()], [0, yb.max()], \"k--\")\n",
        "ax[1].set_title(\"Monthly failure count: actual vs predicted\"); ax[1].legend()\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 추천 엔진 검증 3890건: top-1 정확도 83.9% (단일 코드 Baseline 68.7%), top-3 포함률 99.5%\n",
        "\n",
        "[신뢰도 보정] 구간 | 건수 | 평균 신뢰도 | 실제 정확도\n",
        " ~0.5       155  0.46  0.48\n",
        " 0.5~0.7   1123  0.60  0.61\n",
        " 0.7~0.9    387  0.78  0.74\n",
        " 0.9~      2225  0.99  1.00\n",
        "기대 보정 오차(ECE) 0.014\n",
        "\n",
        "[진단 카드 예시] 정비 조치 가이드라인\n",
        " 입력 DTC ['G01', 'C101', 'S01'] -> 연료분사기 95%, 터보차저 5%, 연료 필터 0%\n",
        "   조치: 연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체\n",
        " 입력 DTC ['G01'] -> 연료분사기 55%, 터보차저 34%, 연료 필터 5%\n",
        "   조치: 연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체  (신뢰도 낮음: 숙련 정비관 확인 권고)\n",
        "\n",
        "[월간 고장 건수 예측] 검증 40건(상위 8개 부품 x 5개월), 선택 하이퍼파라미터 ridge alpha=10, GBR depth=3\n",
        " 기준(전월값)  MAE 15.50  RMSE 21.96  R2 0.860\n",
        " 고도화(앙상블) MAE 10.62  RMSE 14.47  R2 0.939\n",
        "개선율 MAE 31.5%, RMSE 34.1% (목표 20% 이상: 달성), R2 0.939 (목표 0.85 이상: 달성)\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 06. 정비병용 모바일/웹 고장 코드 진단 및 정비 조치 절차 대시보드 구현\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "정비병용 모바일/웹 고장 코드 진단 대시보드를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **정비병용 고장 코드 진단 화면(진단 카드와 웹 대시보드 구성)을 구현하고, 예측마다 기여도 상위 3개 인자와 자연어 요약을 자동 산출하는 설명가능성(XAI) 모듈을 만든다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **모델의 입력 인자별 기여도를 계산해 건마다 상위 3개 근거를 뽑고, 인자를 평균값으로 되돌렸을 때 예측 확률이 낮아지는지로 근거 유효성을 점검 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/explainability/mnt03_xai_explainer.py 및 XAI 리포트` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 군수 장교/군무원이 신뢰하고 조치할 수 있는 설명의 명확성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 주요 예측 건당 기여도 상위 3개 인자 자동 산출률 100%\n",
        "\n",
        "\n",
        "- 진행 기준: SHAP summary plot 및 waterfall plot 생성 파이프라인 완성\n",
        "\n",
        "\n",
        "- 검증 방법: 실무자 대상 설명 근거 유용성 평가 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"정비병이 모바일/웹 화면에서 DTC를 입력하면 후보 고장 부품·신뢰도·정비 조치 절차와 판단 근거(기여도 상위 3개 인자)를 보여 주는 진단 대시보드를 구현한다. 모든 예측 건에 상위 3개 인자를 자동 산출하고 근거 제거 시 확률이 실제로 떨어지는지 점검하며, 자연어 요약과 경보 현황 화면을 만든다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from sklearn.linear_model import LogisticRegression\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 부품별 정비 조치 가이드라인(요약 3단계). 실제 정비 교범이 아니라 교육용으로 만든 예시 문구다.\n",
        "# ---------------------------------------------------------------------------\n",
        "GUIDE = {\n",
        "    \"P01\": \"연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체\", \"P02\": \"흡기 압력·이물 점검 -> 축 유격 확인 -> 불량 시 터보차저 교체\",\n",
        "    \"P03\": \"단자 전압·부식 점검 -> 충전 계통 확인 -> 불량 시 축전지 교체\", \"P04\": \"발전 전압 측정 -> 벨트·배선 점검 -> 불량 시 발전기 교체\",\n",
        "    \"P05\": \"제어모듈 소프트웨어·커넥터 점검 -> 변속 이력 확인 -> 불량 시 모듈 교체\", \"P06\": \"냉각수량·누수 점검 -> 펌프 작동음 확인 -> 불량 시 펌프 교체\",\n",
        "    \"P07\": \"유압 오일량·압력 점검 -> 펌프 누설 확인 -> 불량 시 펌프 교체\", \"P08\": \"센서 배선·간극 점검 -> 신호 파형 확인 -> 불량 시 센서 교체\",\n",
        "    \"P09\": \"커넥터 접촉·부식 점검 -> 핀 상태 확인 -> 불량 시 커넥터 교체\", \"P10\": \"필터 오염도 점검 -> 연료 흐름 확인 -> 불량 시 필터 교체\",\n",
        "    \"P11\": \"제동 압력 점검 -> 밸브 작동 확인 -> 불량 시 밸브 교체(안전 확인 필수)\", \"P12\": \"조향 반응 점검 -> 액추에이터 작동 확인 -> 불량 시 교체(안전 확인 필수)\",\n",
        "    \"P13\": \"필터 차압 점검 -> 오염 확인 -> 불량 시 필터 교체\", \"P14\": \"통신 링크 상태 점검 -> 모듈 전원 확인 -> 불량 시 모듈 교체\",\n",
        "    \"P15\": \"압력 센서 신호 점검 -> 윤활유량 확인 -> 불량 시 센서 교체\", \"P16\": \"팬 모터 전류 점검 -> 회전 확인 -> 불량 시 모터 교체\"}\n",
        "\n",
        "SYSKR = {\"engine\": \"엔진\", \"electric\": \"전기\", \"drive\": \"구동\", \"hydraulic\": \"유압\", \"brake\": \"제동\", \"cooling\": \"냉각\"}\n",
        "SHNAME = {SHARED[p[0]]: f\"{PNAME[p[0]]}·{PNAME[p[1]]} 공유\" for p in PAIRS}\n",
        "\n",
        "\n",
        "def code_name(c):\n",
        "    if c.startswith(\"S\"):\n",
        "        return f\"공유 코드 {c}({SHNAME[c]})\"\n",
        "    if c.startswith(\"G\"):\n",
        "        return f\"{SYSKR[[s for s, v in SYMP.items() if v == c][0]]} 계통 증상 코드 {c}\"\n",
        "    return f\"{PNAME['P' + c[1:3]]} 고유 코드 {c}\"\n",
        "\n",
        "\n",
        "FNAME = [code_name(c) for c in CODES] + [\"동시 발생 코드 수\", \"차령\", \"예방정비 경과시간\", \"지형 가혹도\", \"운용 부하\", \"기온\"]\n",
        "NC = len(CODES)\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "train, test = ev[ev[\"month\"] < SPLIT_M], ev[ev[\"month\"] >= SPLIT_M].reset_index(drop=True)\n",
        "Xtr, Xte = feats(train, fleet), feats(test, fleet)\n",
        "clf = LogisticRegression(C=3, max_iter=500).fit(Xtr, train[\"part\"])\n",
        "P = clf.predict_proba(Xte)\n",
        "cls, mu = clf.classes_, Xtr.mean(axis=0)\n",
        "pred = P.argmax(axis=1)\n",
        "\n",
        "\n",
        "def explain(i, k=3):\n",
        "    \"\"\"예측 부품에 대한 입력 인자별 기여도 = 계수 x (입력값 - 학습 평균). 값이 큰 인자 k개를 반환.\"\"\"\n",
        "    contrib = clf.coef_[pred[i]] * (Xte[i] - mu)\n",
        "    contrib[:NC][Xte[i][:NC] == 0] = -np.inf                  # 입력되지 않은 코드는 근거에서 제외\n",
        "    idx = np.argsort(-contrib)[:k]\n",
        "    return [(FNAME[j], contrib[j]) for j in idx], idx\n",
        "\n",
        "\n",
        "def summary(i):\n",
        "    f, _ = explain(i)\n",
        "    return (f\"{PNAME[cls[pred[i]]]} 고장 가능성 {P[i, pred[i]]:.0%}. 주요 근거: \" + \", \".join(n for n, _ in f) + \".\")\n",
        "\n",
        "\n",
        "# 1) 전 예측 건 기여도 자동 산출 + 근거 유효성 점검\n",
        "ok3, drop = 0, 0\n",
        "for i in range(len(test)):\n",
        "    f, idx = explain(i)\n",
        "    ok3 += len(f) == 3\n",
        "    x2 = Xte[i].copy(); x2[idx[0]] = mu[idx[0]]              # 가장 큰 근거를 평균값으로 되돌려 본다\n",
        "    drop += clf.predict_proba(x2[None])[0, pred[i]] < P[i, pred[i]]\n",
        "print(f\"[교육용가상자료] 예측 {len(test)}건 중 기여도 상위 3개 인자 산출 {ok3}건 (산출률 {ok3 / len(test):.0%}, 목표 100%: {'달성' if ok3 == len(test) else '미달'})\")\n",
        "print(f\"근거 유효성: 상위 인자를 평균값으로 되돌리면 예측 확률이 낮아지는 비율 {drop / len(test):.1%}\")\n",
        "\n",
        "# 2) 모바일 진단 카드 2건: 신뢰도 높음 / 낮음\n",
        "print(\"\\n[모바일 진단 카드]\")\n",
        "for tgt in (0.95, 0.55):\n",
        "    i = int(np.argmin(np.abs(P.max(axis=1) - tgt)))\n",
        "    f, _ = explain(i)\n",
        "    print(f\" 입력 DTC {test['codes'][i]}  (실제 고장 {PNAME[test['part'][i]]})\")\n",
        "    print(f\"   진단 {PNAME[cls[pred[i]]]} 신뢰도 {P[i, pred[i]]:.0%} | 근거: \" + \" / \".join(f\"{n}({v:+.1f})\" for n, v in f))\n",
        "    print(f\"   조치 절차: {GUIDE[cls[pred[i]]]}\")\n",
        "    print(f\"   요약: {summary(i)}\")\n",
        "\n",
        "# 3) 웹 대시보드: 최근 달 현황과 경보\n",
        "last = test[test[\"month\"] == MONTHS - 1]\n",
        "li = last.index.values\n",
        "alert = [(PSYS[cls[pred[i]]] in CRIT_SYS) and P[i, pred[i]] >= 0.7 for i in li]\n",
        "print(f\"\\n[웹 대시보드] 최근 달 정비 건 {len(last)}건, 기동·제동 계통 경보 {sum(alert)}건, 신뢰도 낮음(0.7 미만) {int((P[li].max(axis=1) < 0.7).sum())}건\")\n",
        "cnt = pd.Series([cls[pred[i]] for i in li]).value_counts().head(3)\n",
        "print(\"  다빈도 진단 부품: \" + \", \".join(f\"{PNAME[p]} {n}건\" for p, n in cnt.items()))\n",
        "\n",
        "fig, ax = plt.subplots(2, 2, figsize=(11, 7))\n",
        "cnt5 = pd.Series([cls[pred[i]] for i in li]).value_counts().head(8)\n",
        "ax[0, 0].bar(cnt5.index, cnt5.values); ax[0, 0].set_title(\"Diagnosed parts (last month)\")\n",
        "mon = ev.groupby(\"month\").size(); ax[0, 1].plot(mon.index, mon.values, marker=\"o\"); ax[0, 1].set_title(\"Monthly maintenance cases\")\n",
        "j = int(np.argmin(np.abs(P.max(axis=1) - 0.55))); f, _ = explain(j)\n",
        "ax[1, 0].barh([f\"factor {k + 1}\" for k in range(3)][::-1], [v for _, v in f][::-1]); ax[1, 0].set_title(\"Top-3 contributions (low-confidence case)\")\n",
        "ax[1, 1].hist(P.max(axis=1), bins=20); ax[1, 1].axvline(0.7, ls=\"--\", color=\"k\"); ax[1, 1].set_title(\"Confidence distribution (line = 0.7)\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 예측 3890건 중 기여도 상위 3개 인자 산출 3890건 (산출률 100%, 목표 100%: 달성)\n",
        "근거 유효성: 상위 인자를 평균값으로 되돌리면 예측 확률이 낮아지는 비율 92.5%\n",
        "\n",
        "[모바일 진단 카드]\n",
        " 입력 DTC ['G01', 'C101', 'S01']  (실제 고장 연료분사기)\n",
        "   진단 연료분사기 신뢰도 95% | 근거: 공유 코드 S01(연료분사기·터보차저 공유)(+4.3) / 엔진 계통 증상 코드 G01(+3.0) / 연료 필터 고유 코드 C101(+1.6)\n",
        "   조치 절차: 연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체\n",
        "   요약: 연료분사기 고장 가능성 95%. 주요 근거: 공유 코드 S01(연료분사기·터보차저 공유), 엔진 계통 증상 코드 G01, 연료 필터 고유 코드 C101.\n",
        " 입력 DTC ['G01']  (실제 고장 터보차저)\n",
        "   진단 연료분사기 신뢰도 55% | 근거: 엔진 계통 증상 코드 G01(+3.0) / 기온(+0.1) / 운용 부하(+0.1)\n",
        "   조치 절차: 연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체\n",
        "   요약: 연료분사기 고장 가능성 55%. 주요 근거: 엔진 계통 증상 코드 G01, 기온, 운용 부하.\n",
        "\n",
        "[웹 대시보드] 최근 달 정비 건 457건, 기동·제동 계통 경보 238건, 신뢰도 낮음(0.7 미만) 148건\n",
        "  다빈도 진단 부품: 연료분사기 189건, 터보차저 72건, 냉각수 펌프 40건\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 07. 복합 고장 의심 시 숙련 정비관 2차 정밀 진단(HITL) 워크플로우 설계\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "복합 고장 의심 시 숙련 정비관 2차 정밀 진단(HITL) 워크플로우를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **복합 고장이 의심될 때 숙련 정비관 2차 정밀 진단(HITL) 워크플로우를 시뮬레이션으로 설계하고 오진단 비용 절감 효과를 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **신뢰도가 기준 미만이거나 3개 계통 이상 DTC가 겹친 건을 정비관 대기열에 올리고, 엔진 안을 자동 확정하지 않은 채 정비관의 확정·정정·무응답(엔진 안 유지, 재확인 표시)을 사유와 함께 해시 체인 감사 로그로 남긴다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 5개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/optimization/mnt03_simulator.py` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 현실적 군수 제약조건 반영도 및 시나리오별 대안 제시 적절성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 최적화 적용 시 비용 15% 절감 또는 재고 회전율 20% 향상\n",
        "\n",
        "\n",
        "- 진행 기준: 선형/정수 계획법(LP/MIP) 최적화 해 도출 및 시뮬레이션 완료\n",
        "\n",
        "\n",
        "- 검증 방법: 과거 실적 데이터 대비 최적화 결과 비교 백테스팅\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"복합 고장이 의심되는 정비 건(신뢰도 낮음, 서로 다른 계통의 DTC 동시 발생)을 숙련 정비관 2차 정밀 진단 대기열에 올려 승인·정정·무응답(엔진 안 유지·재확인 표시)을 사유와 함께 해시 체인 감사 로그에 남기는 HITL 워크플로우 시뮬레이터를 구현한다. 자동 확정 0건, 오진단 비용 절감률, 신뢰도 기준·정비관 수용량·고장 급증에 대한 민감도를 확인한다.\"\"\"\n",
        "import hashlib\n",
        "import json\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from sklearn.linear_model import LogisticRegression\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "train, test = ev[ev[\"month\"] < SPLIT_M], ev[ev[\"month\"] >= SPLIT_M].reset_index(drop=True)\n",
        "clf = LogisticRegression(C=3, max_iter=500).fit(feats(train, fleet), train[\"part\"])\n",
        "P = clf.predict_proba(feats(test, fleet))\n",
        "cls = clf.classes_\n",
        "top1 = cls[P.argmax(axis=1)]\n",
        "conf = P.max(axis=1)\n",
        "CSYS = {c: {PSYS[p] for p in PID if c in (SPEC[p], SHARED.get(p)) or c == SYMP[PSYS[p]]} for c in CODES}\n",
        "nsys = test[\"codes\"].map(lambda cs: len(set().union(*[CSYS[c] for c in cs])))\n",
        "multi_sys = ((test[\"codes\"].str.len() >= 3) & (nsys >= 3)).values          # 3개 이상 계통에 걸친 DTC가 동시에 발생\n",
        "THR, CAP = 0.8, 300      # 운영점: 신뢰도 0.8 미만 또는 다계통 동시 발생이면 정밀 진단, 정비관 월 수용량 300건(아래 민감도 분석으로 선정)\n",
        "\n",
        "# [교육용가상자료] 가정 비용지수: 오진단 시 헛교체 부품비 + 헛교체 공수 + 재입고 패널티 3.0 + 올바른 수리, 정밀 진단 1건 1.0\n",
        "REVIEW_COST, REVISIT, SENIOR_ACC, NO_RESP = 1.0, 3.0, 0.93, 0.08\n",
        "\n",
        "\n",
        "def case_cost(pred, true):\n",
        "    if pred == true:\n",
        "        return REPH[true]\n",
        "    return COST[pred] + REPH[pred] + REVISIT + REPH[true]\n",
        "\n",
        "\n",
        "class Audit:\n",
        "    def __init__(self):\n",
        "        self.log = []\n",
        "\n",
        "    def add(self, **body):\n",
        "        prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
        "        body = {\"seq\": len(self.log), **body}\n",
        "        self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, sort_keys=True, ensure_ascii=False)).encode()).hexdigest()})\n",
        "\n",
        "    def verify(self):\n",
        "        prev = \"GENESIS\"\n",
        "        for e in self.log:\n",
        "            b = {k: v for k, v in e.items() if k not in (\"prev\", \"hash\")}\n",
        "            if e[\"prev\"] != prev or hashlib.sha256((prev + json.dumps(b, sort_keys=True, ensure_ascii=False)).encode()).hexdigest() != e[\"hash\"]:\n",
        "                return e[\"seq\"]\n",
        "            prev = e[\"hash\"]\n",
        "        return None\n",
        "\n",
        "\n",
        "def run(thr=THR, cap=CAP, surge=1.0, seed=0, audit=None, mode=\"hitl\"):\n",
        "    \"\"\"mode: auto(엔진 top-1 전부 확정: 비교용, 실제 운용 금지) / hitl(복합 의심 건은 정비관 결정 전까지 확정하지 않음)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    idx_all = np.arange(len(test))\n",
        "    if surge != 1.0:\n",
        "        idx_all = np.concatenate([idx_all, rng.choice(idx_all, int(len(idx_all) * (surge - 1)), replace=True)])\n",
        "    cost, wrong, queued, decided, pending_auto, silent = 0.0, 0, 0, 0, 0, 0\n",
        "    for m in range(SPLIT_M, MONTHS):\n",
        "        mi = idx_all[test[\"month\"].values[idx_all] == m]\n",
        "        flag = [i for i in mi if conf[i] < thr or multi_sys[i]]\n",
        "        flag.sort(key=lambda i: (PSYS[top1[i]] not in CRIT_SYS, conf[i]))      # 안전 계통 우선, 신뢰도 낮은 순\n",
        "        queue = set(flag[:cap]) if mode == \"hitl\" else set()\n",
        "        for i in mi:\n",
        "            true, final = test[\"part\"][i], top1[i]\n",
        "            if i in queue:\n",
        "                queued += 1\n",
        "                audit.add(event=\"PROPOSE\", case=int(i), engine=str(top1[i]), conf=round(float(conf[i]), 3), status=\"PENDING\")\n",
        "                cost += REVIEW_COST\n",
        "                r = rng.random()\n",
        "                if r < NO_RESP:\n",
        "                    dec, why = \"무응답\", \"응답 지연 -> 엔진 안 유지, 재확인 필요 표시\"; silent += 1\n",
        "                elif r < NO_RESP + (1 - NO_RESP) * SENIOR_ACC:\n",
        "                    dec, final, why = (\"승인\" if top1[i] == true else \"정정\"), true, \"추가 계측·부품 분리 점검으로 확인\"\n",
        "                else:\n",
        "                    second = cls[np.argsort(-P[i])[1]]\n",
        "                    dec, final, why = \"정정\", second, \"2순위 후보로 정정(근거 부족)\"\n",
        "                audit.add(event=\"DECIDE\", case=int(i), decision=dec, final=str(final), reason=why, by=\"숙련 정비관(모의)\")\n",
        "                decided += 1\n",
        "            cost += case_cost(final, true); wrong += final != true\n",
        "    n = len(idx_all)\n",
        "    return {\"cost\": cost, \"wrong\": wrong / n, \"queued\": queued, \"decided\": decided, \"silent\": silent, \"n\": n, \"auto\": pending_auto}\n",
        "\n",
        "\n",
        "aud = Audit()\n",
        "A, H = run(mode=\"auto\", audit=Audit()), run(audit=aud)\n",
        "print(f\"[교육용가상자료] 검증 정비 건 {A['n']}건, 복합 고장 의심 건 {int(((conf < THR) | multi_sys).sum())}건 (신뢰도 {THR} 미만 또는 3개 계통 이상 DTC)\")\n",
        "print(f\"엔진 단독 자동 확정(비교용): 비용지수 {A['cost']:.0f}, 오진단율 {A['wrong']:.1%}\")\n",
        "print(f\"HITL(정비관 대기열 {H['queued']}건): 비용지수 {H['cost']:.0f}, 오진단율 {H['wrong']:.1%}\")\n",
        "cut = 1 - H[\"cost\"] / A[\"cost\"]\n",
        "print(f\"오진단 비용 절감률 {cut:.1%} (목표 15% 이상: {'달성' if cut >= 0.15 else '미달'})\")\n",
        "\n",
        "# 승인 체계 점검\n",
        "print(f\"\\n자동 확정 0건 확인: 대기열 건은 모두 정비관 결정 기록 후 확정 (자동 확정 {H['auto']}건)\")\n",
        "print(f\"결정 로깅률 {H['decided'] / max(H['queued'], 1):.0%} ({H['decided']}/{H['queued']}), 무응답 {H['silent']}건은 엔진 안 유지 + 재확인 표시\")\n",
        "print(f\"감사 로그 무결성: {'정상' if aud.verify() is None else '이상'}\", end=\", \")\n",
        "aud.log[7][\"reason\"] = \"사후 수정\"\n",
        "print(f\"사후 변조 시 seq {aud.verify()}에서 탐지\")\n",
        "print(\"결정 사유 예:\", \"; \".join(sorted({e[\"reason\"] for e in aud.log if e[\"event\"] == \"DECIDE\"})[:3]))\n",
        "\n",
        "# 민감도 분석\n",
        "print(\"\\n[민감도 1] 신뢰도 기준(미만이면 정밀 진단) -> 절감률, 정비관 대기열(6개월)\")\n",
        "rows = []\n",
        "for thr in (0.5, 0.6, 0.7, 0.8, 0.9):\n",
        "    r = run(thr=thr, cap=10 ** 6, audit=Audit()); rows.append((thr, 1 - r[\"cost\"] / A[\"cost\"], r[\"queued\"]))\n",
        "    print(f\"  {thr:.1f}: 절감 {rows[-1][1]:.1%}, 대기열 {r['queued']}건\")\n",
        "print(\"[민감도 2] 정비관 월 수용량 -> 절감률 (전시 전환 가정: 고장 1.5배 급증)\")\n",
        "for cap in (50, 100, 200, 300):\n",
        "    r1, r2 = run(cap=cap, audit=Audit()), run(cap=cap, surge=1.5, audit=Audit())\n",
        "    b2 = run(mode=\"auto\", surge=1.5, audit=Audit())\n",
        "    print(f\"  월 {cap}건: 평시 {1 - r1['cost'] / A['cost']:.1%}, 급증 {1 - r2['cost'] / b2['cost']:.1%}\")\n",
        "\n",
        "plt.figure(figsize=(6, 3.5))\n",
        "plt.plot([r[0] for r in rows], [r[1] * 100 for r in rows], marker=\"o\"); plt.xlabel(\"confidence threshold\"); plt.ylabel(\"cost saving (%)\")\n",
        "plt.title(\"HITL sensitivity: confidence threshold\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 검증 정비 건 3890건, 복합 고장 의심 건 1555건 (신뢰도 0.8 미만 또는 3개 계통 이상 DTC)\n",
        "엔진 단독 자동 확정(비교용): 비용지수 17546, 오진단율 16.1%\n",
        "HITL(정비관 대기열 1501건): 비용지수 14809, 오진단율 4.2%\n",
        "오진단 비용 절감률 15.6% (목표 15% 이상: 달성)\n",
        "\n",
        "자동 확정 0건 확인: 대기열 건은 모두 정비관 결정 기록 후 확정 (자동 확정 0건)\n",
        "결정 로깅률 100% (1501/1501), 무응답 120건은 엔진 안 유지 + 재확인 표시\n",
        "감사 로그 무결성: 정상, 사후 변조 시 seq 7에서 탐지\n",
        "결정 사유 예: 2순위 후보로 정정(근거 부족); 사후 수정; 응답 지연 -> 엔진 안 유지, 재확인 필요 표시\n",
        "\n",
        "[민감도 1] 신뢰도 기준(미만이면 정밀 진단) -> 절감률, 정비관 대기열(6개월)\n",
        "  0.5: 절감 1.9%, 대기열 162건\n",
        "  0.6: 절감 9.4%, 대기열 784건\n",
        "  0.7: 절감 14.6%, 대기열 1285건\n",
        "  0.8: 절감 16.9%, 대기열 1555건\n",
        "  0.9: 절감 17.7%, 대기열 1671건\n",
        "[민감도 2] 정비관 월 수용량 -> 절감률 (전시 전환 가정: 고장 1.5배 급증)\n",
        "  월 50건: 평시 2.8%, 급증 2.8%\n",
        "  월 100건: 평시 7.0%, 급증 4.4%\n",
        "  월 200건: 평시 13.8%, 급증 9.3%\n",
        "  월 300건: 평시 15.6%, 급증 12.7%\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 08. 1,000건 동시 다발 고장 코드 주입 시 연관 고장 추론 부하 테스트\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "1,000건 동시 고장 코드 주입 시 연관 고장 추론 부하 테스트를 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **1,000건의 DTC 진단 요청을 동시에 주입해 연관 고장 추론의 처리 시간과 경보 정확도를 확인하는 부하 테스트를 수행한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **가상 진단 요청 1,000건에 표기 오류, 빈 요청, 미정의 코드, 미등록 차량, 중복 코드 같은 오류를 섞어 일괄 처리하고, 정상 처리·격리 건수와 유실 여부를 센다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/dashboard/mnt03_app.py (Streamlit/Dash 등)` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 의사결정 지원에 최적화된 시각적 위계 및 사용자 경험(UX)\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 대시보드 로딩 시간 2초 이내, 경보 발령 정확도 95% 이상\n",
        "\n",
        "\n",
        "- 진행 기준: 웹 기반 분석 대시보드 구동 및 실시간 필터링 기능 검증\n",
        "\n",
        "\n",
        "- 검증 방법: 실제 군수 데이터 연동 화면 테스트 및 UI 동작 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"1,000건의 DTC 진단 요청을 동시에 주입해 연관 고장 추론(부품 진단 + 기동·제동 계통 경보)의 처리 시간과 경보 정확도를 측정하는 부하 테스트를 수행한다. 형식이 깨진 코드·빈 요청·미등록 차량을 섞은 오류 주입 요청이 격리 처리되는지, 요청 수를 100~5,000건으로 늘렸을 때 시간이 어떻게 변하는지 확인한다.\"\"\"\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from concurrent.futures import ThreadPoolExecutor\n",
        "from sklearn.linear_model import LogisticRegression\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "train, test = ev[ev[\"month\"] < SPLIT_M], ev[ev[\"month\"] >= SPLIT_M].reset_index(drop=True)\n",
        "clf = LogisticRegression(C=3, max_iter=500).fit(feats(train, fleet), train[\"part\"])\n",
        "cls, FL = clf.classes_, fleet.set_index(\"veh\")\n",
        "VALID = set(CODES)\n",
        "\n",
        "\n",
        "def make_requests(n, seed=0, bad_rate=0.05):\n",
        "    \"\"\"[교육용가상자료] 검증 기간 정비 건에서 n건을 뽑아 DTC 요청으로 만든다. bad_rate 비율만큼 오류를 주입한다.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    rows = test.iloc[rng.choice(len(test), n, replace=n > len(test))]\n",
        "    reqs, kinds = [], []\n",
        "    for e in rows.itertuples():\n",
        "        r, kind = {\"veh\": e.veh, \"codes\": list(e.codes), \"temp\": e.temp, \"true\": e.part}, \"정상\"\n",
        "        u = rng.random()\n",
        "        if u < bad_rate * 0.3: r[\"codes\"] = [c.lower() + \" \" for c in r[\"codes\"]]; kind = \"표기 오류\"\n",
        "        elif u < bad_rate * 0.5: r[\"codes\"] = [\"ZZ9\"]; kind = \"미정의 코드\"\n",
        "        elif u < bad_rate * 0.7: r[\"codes\"] = []; kind = \"빈 요청\"\n",
        "        elif u < bad_rate * 0.85: r[\"veh\"] = \"V999\"; kind = \"미등록 차량\"\n",
        "        elif u < bad_rate: r[\"codes\"] = r[\"codes\"] * 3; kind = \"중복 코드\"\n",
        "        reqs.append(r); kinds.append(kind)\n",
        "    return reqs, kinds\n",
        "\n",
        "\n",
        "def normalize(r):\n",
        "    \"\"\"표기 정리(공백·대소문자) 후 검증. 문제가 있으면 예외 대신 사유를 돌려준다.\"\"\"\n",
        "    codes = sorted({c.strip().upper() for c in r[\"codes\"]})\n",
        "    if r[\"veh\"] not in FL.index: return None, \"미등록 차량\"\n",
        "    if not codes or any(c not in VALID for c in codes): return None, \"빈 요청·미정의 코드\"\n",
        "    return codes, \"\"\n",
        "\n",
        "\n",
        "def build_x(batch):\n",
        "    X = np.zeros((len(batch), len(CODES) + 6))\n",
        "    for i, (r, cs) in enumerate(batch):\n",
        "        for c in cs: X[i, CIDX[c]] = 1\n",
        "        v = FL.loc[r[\"veh\"]]\n",
        "        X[i, len(CODES):] = [len(cs) / 3, v[\"age\"] / 25, v[\"pm_gap\"] / 200, v[\"terrain\"], v[\"load\"], (r[\"temp\"] - 12) / 15]\n",
        "    return X\n",
        "\n",
        "\n",
        "def serve_batch(reqs):\n",
        "    ok, rej = [], []\n",
        "    for k, r in enumerate(reqs):\n",
        "        cs, why = normalize(r)\n",
        "        (ok if cs else rej).append((k, r, cs) if cs else (k, why))\n",
        "    X = build_x([(r, cs) for _, r, cs in ok])\n",
        "    P = clf.predict_proba(X) if len(ok) else np.zeros((0, len(cls)))\n",
        "    top = cls[P.argmax(axis=1)] if len(ok) else []\n",
        "    out = [{\"k\": k, \"part\": p, \"conf\": float(P[i].max()), \"alert\": PSYS[p] in CRIT_SYS} for i, ((k, _, _), p) in enumerate(zip(ok, top))]\n",
        "    return out, rej\n",
        "\n",
        "\n",
        "def serve_one(r):\n",
        "    cs, why = normalize(r)\n",
        "    if not cs: return None\n",
        "    p = cls[clf.predict_proba(build_x([(r, cs)]))[0].argmax()]\n",
        "    return {\"part\": p, \"alert\": PSYS[p] in CRIT_SYS}\n",
        "\n",
        "\n",
        "# 1) 1,000건 동시 주입\n",
        "reqs, kinds = make_requests(1000)\n",
        "t0 = time.perf_counter(); out, rej = serve_batch(reqs)\n",
        "agg = pd.DataFrame(out).groupby(\"part\").size().sort_values(ascending=False).head(5)          # 대시보드 집계까지 포함\n",
        "alerts = sum(o[\"alert\"] for o in out); dt = time.perf_counter() - t0\n",
        "print(f\"[교육용가상자료] 1,000건 동시 주입: 정상 처리 {len(out)}건, 격리(오류) {len(rej)}건, 처리 시간 {dt:.2f}초 (목표 2초 이내: {'달성' if dt <= 2 else '미달'})\")\n",
        "print(f\"  오류 주입 {sum(k != '정상' for k in kinds)}건 유형: \" + \", \".join(f\"{k} {v}\" for k, v in pd.Series(kinds).value_counts().items() if k != \"정상\"))\n",
        "print(\"  격리 사유: \" + \", \".join(f\"{k} {v}건\" for k, v in pd.Series([w for _, w in rej]).value_counts().items()))\n",
        "\n",
        "# 2) 경보 정확도(기동·제동 계통 고장 경보)\n",
        "truth = np.array([PSYS[reqs[o[\"k\"]][\"true\"]] in CRIT_SYS for o in out]); pa = np.array([o[\"alert\"] for o in out])\n",
        "acc = (truth == pa).mean(); rec = (pa & truth).sum() / truth.sum(); prec = (pa & truth).sum() / pa.sum()\n",
        "part_acc = np.mean([o[\"part\"] == reqs[o[\"k\"]][\"true\"] for o in out])\n",
        "print(f\"\\n경보 정확도 {acc:.1%} (목표 95% 이상: {'달성' if acc >= 0.95 else '미달'}), 재현율 {rec:.1%}, 정밀도 {prec:.1%}, 경보 {alerts}건\")\n",
        "print(f\"참고: 부품 단위 진단 정확도 {part_acc:.1%}, 상위 부품 \" + \", \".join(f\"{PNAME[p]} {n}\" for p, n in agg.items()))\n",
        "\n",
        "# 3) 단건 호출 방식과 비교(8개 작업자 동시 호출)\n",
        "t0 = time.perf_counter()\n",
        "with ThreadPoolExecutor(8) as ex:\n",
        "    res = list(ex.map(serve_one, reqs))\n",
        "dt1 = time.perf_counter() - t0\n",
        "print(f\"\\n단건 호출 8스레드 {dt1:.2f}초 vs 일괄 처리 {dt:.2f}초 (일괄이 {dt1 / dt:.1f}배 빠름), 단건 방식 예외 0건: {all(r is None or 'part' in r for r in res)}\")\n",
        "\n",
        "# 4) 요청 수 확장 시험\n",
        "print(\"[요청 수별 일괄 처리 시간]\")\n",
        "ns, ts = [100, 500, 1000, 2000, 5000], []\n",
        "for n in ns:\n",
        "    rq, _ = make_requests(n, seed=n)\n",
        "    t0 = time.perf_counter(); serve_batch(rq); ts.append(time.perf_counter() - t0)\n",
        "    print(f\"  {n:5d}건: {ts[-1]:.2f}초 ({n / ts[-1]:.0f}건/초)\")\n",
        "\n",
        "plt.figure(figsize=(6, 3.5)); plt.plot(ns, ts, marker=\"o\"); plt.axhline(2, ls=\"--\", color=\"k\")\n",
        "plt.xlabel(\"requests\"); plt.ylabel(\"seconds\"); plt.title(\"Batch inference time (line = 2 s)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 1,000건 동시 주입: 정상 처리 965건, 격리(오류) 35건, 처리 시간 0.11초 (목표 2초 이내: 달성)\n",
        "  오류 주입 65건 유형: 표기 오류 23, 빈 요청 15, 미정의 코드 11, 미등록 차량 9, 중복 코드 7\n",
        "  격리 사유: 빈 요청·미정의 코드 26건, 미등록 차량 9건\n",
        "\n",
        "경보 정확도 99.8% (목표 95% 이상: 달성), 재현율 100.0%, 정밀도 99.6%, 경보 567건\n",
        "참고: 부품 단위 진단 정확도 85.1%, 상위 부품 연료분사기 303, 축전지 218, 터보차저 126, 발전기 119, 변속기 제어모듈 53\n",
        "\n",
        "단건 호출 8스레드 0.33초 vs 일괄 처리 0.11초 (일괄이 3.1배 빠름), 단건 방식 예외 0건: True\n",
        "[요청 수별 일괄 처리 시간]\n",
        "    100건: 0.01초 (15586건/초)\n",
        "    500건: 0.03초 (18195건/초)\n",
        "   1000건: 0.06초 (16478건/초)\n",
        "   2000건: 0.30초 (6703건/초)\n",
        "   5000건: 0.33초 (15293건/초)\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 09. AAR 기반 연관 고장 오진단 사례 분석 및 고장 룰베이스 재튜닝\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "AAR 기반 연관 고장 오진단 사례를 분석하고 룰베이스를 재튜닝한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **AAR 기간의 연관 고장 오진단 사례를 원인별로 분석하고 고장 룰베이스(코드 유형 가중치·예외 규칙)를 재튜닝해 효과를 지표로 산출한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **AAR 기간 오진단 건을 모호 코드 입력, 연쇄 고장 코드 오인, 복수 고유 코드 충돌 등으로 분류하고 대표 사례 정리하고 규격화하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `reports/mnt03_roi_analysis.md` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 추정 산식의 객관성 및 군 재정/군수 기획 부서 설득력\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 비용 절감액, 시간 단축률 등 3대 정량 효과 지표 산출\n",
        "\n",
        "\n",
        "- 진행 기준: ROI 분석 보고서 완성 및 주요 지표 민감도 분석 수록\n",
        "\n",
        "\n",
        "- 검증 방법: 회계/기획 기준 모델과의 비교 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"AAR(사후검토) 대상 기간의 연관 고장 오진단 사례를 원인별(모호 코드만 입력, 연쇄 고장 코드 오인, 잡음 코드 오인, 기타)로 분류하고, 고장 룰베이스의 코드 유형별 가중치와 사전확률 보정을 AAR 기간 자료로 재튜닝한 뒤 별도 검증 기간에서 효과를 확인한다. 비용 절감액, 재작업 시간 단축률, 헛교체 부품 감소율의 3대 효과 지표와 단계별 도입 로드맵을 산출한다.\"\"\"\n",
        "import itertools\n",
        "from collections import Counter\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "train, aar, val = ev[ev[\"month\"] < 12], ev[(ev[\"month\"] >= 12) & (ev[\"month\"] < 15)].reset_index(drop=True), ev[ev[\"month\"] >= 15].reset_index(drop=True)\n",
        "\n",
        "# 1) 룰베이스: 점수(부품) = sum[코드 유형 가중치 x P(부품 | 코드)] + 사전확률 보정. 학습 기간 조건부확률표를 쓴다.\n",
        "Xtr = multi_hot(train)\n",
        "Y = pd.get_dummies(train[\"part\"]).reindex(columns=PID, fill_value=0).values.astype(float)\n",
        "COND = (Xtr.T @ Y) / np.maximum(Xtr.sum(axis=0)[:, None], 1)             # P(부품 | 코드)\n",
        "PRIOR = np.log(Y.mean(axis=0) + 1e-9)\n",
        "CTYPE = np.array([0 if c in SPEC.values() else 1 if c in SHARED.values() else 2 for c in CODES])    # 0 고유 / 1 공유 / 2 계통 증상\n",
        "V1 = {\"w\": (1.0, 1.0, 1.0), \"beta\": 0.0}                                  # 기존 룰베이스(모든 코드 동일 가중, 사전확률 미반영)\n",
        "\n",
        "\n",
        "def predict(ev_, p):\n",
        "    W = np.array(p[\"w\"])[CTYPE][:, None] * COND\n",
        "    S = multi_hot(ev_) @ W + p[\"beta\"] * PRIOR[None, :]\n",
        "    base = np.array(PID)[S.argmax(axis=1)]\n",
        "    ov = p.get(\"override\", {})                                            # AAR에서 만든 예외 규칙(DTC 조합 -> 부품)이 우선\n",
        "    return np.array([ov.get(frozenset(c), b) for c, b in zip(ev_[\"codes\"], base)])\n",
        "\n",
        "\n",
        "def cost(pred, true):\n",
        "    return sum(REPH[t] if p == t else COST[p] + REPH[p] + 3.0 + REPH[t] for p, t in zip(pred, true))\n",
        "\n",
        "\n",
        "def err_stats(pred, true):\n",
        "    wrong = pred != true.values\n",
        "    return int(wrong.sum()), sum(REPH[p] + 3.0 + REPH[t] for p, t in zip(pred[wrong], true[wrong])), int(wrong.sum())\n",
        "\n",
        "\n",
        "# 2) AAR: 오진단 원인 분류\n",
        "pa = predict(aar, V1)\n",
        "casc = {(p, q) for p, lst in CASCADE.items() for q, _ in lst}\n",
        "cause = []\n",
        "for cs, t, q in zip(aar[\"codes\"], aar[\"part\"], pa):\n",
        "    if t == q: continue\n",
        "    sp = [c for c in cs if c in SPEC.values()]\n",
        "    if not sp: cause.append(\"모호 코드(공유·증상)만 입력\")\n",
        "    elif len(sp) >= 2 and ((t, q) in casc or (q, t) in casc): cause.append(\"연쇄 고장 코드 오인\")\n",
        "    elif len(sp) >= 2: cause.append(\"복수 고유 코드 충돌(잡음 포함)\")\n",
        "    else: cause.append(\"기타(고유 코드 1개, 확률 편향)\")\n",
        "C = pd.Series(cause).value_counts()\n",
        "print(f\"[교육용가상자료] AAR 기간(12~14월) 정비 {len(aar)}건: 기존 룰베이스 오진단 {len(cause)}건 ({len(cause) / len(aar):.1%})\")\n",
        "for k, v in C.items():\n",
        "    print(f\"  {k:24s} {v:4d}건 ({v / len(cause):.0%})\")\n",
        "ex = [(cs, PNAME[t], PNAME[q]) for cs, t, q in zip(aar[\"codes\"], aar[\"part\"], pa) if t != q and sum(c in SPEC.values() for c in cs) >= 2][:1]\n",
        "print(\"  사례:\", f\"입력 DTC {ex[0][0]} -> 오진단 {ex[0][2]}, 실제 {ex[0][1]}\" if ex else \"-\")\n",
        "\n",
        "# 3) 재튜닝: (가) 코드 유형 가중치·사전확률 격자 탐색, (나) AAR 오진단 조합에서 예외 규칙 추출. 모두 AAR 기간 자료만 쓰고 검증 기간은 건드리지 않는다.\n",
        "best = None\n",
        "for w in itertools.product((0.5, 1.0, 1.5, 2.0), repeat=3):\n",
        "    for beta in (0.0, 0.2, 0.5, 1.0):\n",
        "        p = {\"w\": w, \"beta\": beta}\n",
        "        c = cost(predict(aar, p), aar[\"part\"])\n",
        "        if best is None or c < best[0]: best = (c, p)\n",
        "V2 = dict(best[1])\n",
        "pa2 = predict(aar, V2)\n",
        "bad_sets = {frozenset(c) for c, t, q in zip(aar[\"codes\"], aar[\"part\"], pa2) if t != q}\n",
        "grp = {}\n",
        "for c, t in zip(aar[\"codes\"], aar[\"part\"]):\n",
        "    grp.setdefault(frozenset(c), Counter())[t] += 1\n",
        "V2[\"override\"] = {k: v.most_common(1)[0][0] for k, v in grp.items()\n",
        "                  if k in bad_sets and sum(v.values()) >= 3 and v.most_common(1)[0][1] / sum(v.values()) >= 0.6}\n",
        "print(f\"\\n재튜닝 결과: 코드 유형 가중치(고유·공유·증상) {V1['w']} -> {V2['w']}, 사전확률 보정 {V1['beta']} -> {V2['beta']}, AAR 예외 규칙 {len(V2['override'])}개 추가\")\n",
        "rows = []\n",
        "for nm, d in ((\"AAR 기간(튜닝)\", aar), (\"검증 기간(15~17월)\", val)):\n",
        "    p1, p2 = predict(d, V1), predict(d, V2)\n",
        "    rows.append((nm, (p1 == d[\"part\"].values).mean(), (p2 == d[\"part\"].values).mean(), cost(p1, d[\"part\"]), cost(p2, d[\"part\"])))\n",
        "    print(f\" {nm:14s} 정확도 {rows[-1][1]:.1%} -> {rows[-1][2]:.1%}, 비용지수 {rows[-1][3]:.0f} -> {rows[-1][4]:.0f}\")\n",
        "\n",
        "# 4) ROI 3대 지표(검증 기간 3개월, 가정 단가 기준 상대값)\n",
        "p1, p2 = predict(val, V1), predict(val, V2)\n",
        "n1, h1, _ = err_stats(p1, val[\"part\"]); n2, h2, _ = err_stats(p2, val[\"part\"])\n",
        "saved = cost(p1, val[\"part\"]) - cost(p2, val[\"part\"])\n",
        "wasted1 = sum(COST[p] for p, t in zip(p1, val[\"part\"]) if p != t); wasted2 = sum(COST[p] for p, t in zip(p2, val[\"part\"]) if p != t)\n",
        "print(\"\\n[ROI 3대 정량 효과 지표] (검증 3개월, 가정 단가)\")\n",
        "print(f\" 1) 비용 절감: 비용지수 {saved:.0f} 절감 ({saved / cost(p1, val['part']):.1%}), 연환산 약 {saved * 4:.0f}\")\n",
        "print(f\" 2) 시간 단축: 오진단 재작업 {h1:.0f}h -> {h2:.0f}h ({1 - h2 / h1:.1%} 단축)\")\n",
        "print(f\" 3) 헛교체 부품 비용: {wasted1:.0f} -> {wasted2:.0f} ({1 - wasted2 / wasted1:.1%} 감소), 오진단 {n1}건 -> {n2}건\")\n",
        "print(\"[도입 로드맵] 단기(1~3월): 가중치 갱신 + 정비관 확인 병행 / 중기(3~12월): AAR 분기 반복·신규 코드 반영 / 장기(1년~): 군수 정보체계 이력 연동\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 3.5))\n",
        "ax[0].barh([f\"cause {i + 1}\" for i in range(len(C))][::-1], C.values[::-1]); ax[0].set_title(\"Misdiagnosis causes (AAR period)\")\n",
        "ax[1].bar([\"before\", \"after\"], [rows[1][1], rows[1][2]]); ax[1].set_ylim(0.5, 1); ax[1].set_title(\"Validation accuracy\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] AAR 기간(12~14월) 정비 2336건: 기존 룰베이스 오진단 522건 (22.3%)\n",
        "  모호 코드(공유·증상)만 입력          243건 (47%)\n",
        "  기타(고유 코드 1개, 확률 편향)       237건 (45%)\n",
        "  연쇄 고장 코드 오인                26건 (5%)\n",
        "  복수 고유 코드 충돌(잡음 포함)         16건 (3%)\n",
        "  사례: 입력 DTC ['S04', 'G01', 'C081', 'C101'] -> 오진단 연료분사기, 실제 연료 필터\n",
        "\n",
        "재튜닝 결과: 코드 유형 가중치(고유·공유·증상) (1.0, 1.0, 1.0) -> (2.0, 1.0, 0.5), 사전확률 보정 0.0 -> 0.0, AAR 예외 규칙 31개 추가\n",
        " AAR 기간(튜닝)     정확도 77.7% -> 82.9%, 비용지수 11938 -> 10805\n",
        " 검증 기간(15~17월)  정확도 78.1% -> 80.1%, 비용지수 8449 -> 7832\n",
        "\n",
        "[ROI 3대 정량 효과 지표] (검증 3개월, 가정 단가)\n",
        " 1) 비용 절감: 비용지수 616 절감 (7.3%), 연환산 약 2466\n",
        " 2) 시간 단축: 오진단 재작업 3038h -> 2713h (10.7% 단축)\n",
        " 3) 헛교체 부품 비용: 1778 -> 1441 (18.9% 감소), 오진단 341건 -> 309건\n",
        "[도입 로드맵] 단기(1~3월): 가중치 갱신 + 정비관 확인 병행 / 중기(3~12월): AAR 분기 반복·신규 코드 반영 / 장기(1년~): 군수 정보체계 이력 연동\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 10. 무기체계 고장 코드(DTC) 기반 연관 고장 예측 및 정비 가이드 도구 완성\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "무기체계 고장 코드(DTC) 기반 연관 고장 예측 및 정비 가이드 도구를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **DTC 입력부터 연관 고장 예측, 정비 가이드, 승인 대상 선별, 재학습까지 한 파일로 통합하고 자체 점검과 브리핑 요약을 출력하는 도구를 완성한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **월별 정비 실적으로 진단 모델을 학습하고, 신규 실적으로 도전자 모델을 만들어 챔피언과 검증 오차를 비교한 뒤 비슷하거나 나을 때만 교체 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `docs/mnt03_operations_guide.md 및 pipeline/mnt03_retrain.py` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 지속 가능한 운영 가능성 및 국방 군수 정보체계와의 연동 용이성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 자동 재학습 파이프라인 정상 실행, 최종 문서 100% 완비\n",
        "\n",
        "\n",
        "- 진행 기준: 최종 인도물 검수 및 프로젝트 완료 발표자료 완성\n",
        "\n",
        "\n",
        "- 검증 방법: End-to-End 전체 파이프라인 자동화 실행 및 최종 검수\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"DTC 로그 검증·정제, 진단 모델 자동 재학습(챔피언·도전자 비교 후 교체), 고장 코드 기반 연관 고장 진단과 신뢰도 산출, 신뢰도 낮은 건의 정비관 승인 대기 분리, 다빈도 고장품목 월간 소요 예측까지 정비 가이드 도구를 한 파일로 통합 실행하고, 자체 점검·운영 문서 완비 점검·정비 담당 장교 브리핑 요약을 출력한다.\"\"\"\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from sklearn.linear_model import LogisticRegression, Ridge\n",
        "from sklearn.preprocessing import StandardScaler\n",
        "from sklearn.pipeline import make_pipeline\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 차량 100대의 18개월 고장 진단 코드(DTC)·정비 이력 생성기 (실제 부대·장비·정비 실적 없음)\n",
        "# 부품 16종(1000가동시간당 고장률 차등), DTC 28종(부품 고유 코드, 짝 부품 공유 코드, 계통 증상 코드)\n",
        "# 고장 1건(정비 이력)마다 DTC가 1~5개 동시 발생하고, 연쇄 고장 코드와 일시 잡음 코드가 섞인다.\n",
        "# ---------------------------------------------------------------------------\n",
        "PARTS = [  # (ID, 이름, 계통, 1000가동시간당 고장률, 가상 단가지수, 수리시간 h)\n",
        "    (\"P01\", \"연료분사기\", \"engine\", 12.0, 3.0, 2.0), (\"P02\", \"터보차저\", \"engine\", 9.0, 8.0, 5.0),\n",
        "    (\"P03\", \"축전지\", \"electric\", 6.5, 4.0, 1.5), (\"P04\", \"발전기\", \"electric\", 4.0, 9.0, 3.0),\n",
        "    (\"P05\", \"변속기 제어모듈\", \"drive\", 3.0, 12.0, 6.0), (\"P06\", \"냉각수 펌프\", \"engine\", 2.2, 3.5, 3.0),\n",
        "    (\"P07\", \"유압 펌프\", \"hydraulic\", 1.6, 10.0, 4.0), (\"P08\", \"속도 센서\", \"drive\", 1.2, 2.0, 1.5),\n",
        "    (\"P09\", \"ECU 커넥터\", \"electric\", 0.9, 2.5, 1.5), (\"P10\", \"연료 필터\", \"engine\", 0.7, 0.8, 0.7),\n",
        "    (\"P11\", \"제동 밸브\", \"brake\", 0.5, 6.0, 3.0), (\"P12\", \"조향 액추에이터\", \"drive\", 0.4, 9.0, 5.0),\n",
        "    (\"P13\", \"공기 필터\", \"engine\", 0.3, 0.5, 0.5), (\"P14\", \"통신 인터페이스 모듈\", \"electric\", 0.25, 3.0, 1.5),\n",
        "    (\"P15\", \"윤활유 압력 센서\", \"engine\", 0.2, 1.0, 1.0), (\"P16\", \"팬 구동 모터\", \"cooling\", 0.15, 2.0, 2.0)]\n",
        "PID = [p[0] for p in PARTS]\n",
        "PNAME = {p[0]: p[1] for p in PARTS}; PSYS = {p[0]: p[2] for p in PARTS}; RATE = {p[0]: p[3] for p in PARTS}\n",
        "COST = {p[0]: p[4] for p in PARTS}; REPH = {p[0]: p[5] for p in PARTS}\n",
        "PAIRS = [(\"P01\", \"P02\"), (\"P03\", \"P04\"), (\"P05\", \"P08\"), (\"P06\", \"P10\"), (\"P09\", \"P14\"), (\"P13\", \"P15\")]\n",
        "SHARED = {p: f\"S{k + 1:02d}\" for k, pr in enumerate(PAIRS) for p in pr}     # 짝 부품이 함께 쓰는 모호한 코드\n",
        "SPEC = {p: f\"C{p[1:]}1\" for p in PID}                                          # 부품 고유 코드\n",
        "SYS = [\"engine\", \"electric\", \"drive\", \"hydraulic\", \"brake\", \"cooling\"]\n",
        "SYMP = {s: f\"G{i + 1:02d}\" for i, s in enumerate(SYS)}                         # 계통 증상 코드\n",
        "CODES = sorted(set(SPEC.values()) | set(SHARED.values()) | set(SYMP.values()))\n",
        "CASCADE = {\"P01\": [(\"P10\", 0.20)], \"P02\": [(\"P06\", 0.20)], \"P03\": [(\"P04\", 0.25)], \"P04\": [(\"P09\", 0.25)],\n",
        "           \"P05\": [(\"P08\", 0.30)], \"P06\": [(\"P16\", 0.30)], \"P09\": [(\"P14\", 0.20)], \"P12\": [(\"P07\", 0.20)]}\n",
        "CRIT_SYS = {\"engine\", \"drive\", \"brake\"}        # 기동·제동 불능 위험 계통(경보 대상)\n",
        "MONTHS = 18\n",
        "\n",
        "\n",
        "def make_fleet(n=100, seed=0):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    f = pd.DataFrame({\"veh\": [f\"V{i:03d}\" for i in range(n)], \"age\": rng.integers(1, 26, n).astype(float),\n",
        "                      \"terrain\": rng.uniform(0, 1, n), \"load\": rng.uniform(0, 1, n), \"pm_gap\": rng.uniform(20, 200, n),\n",
        "                      \"humid\": rng.uniform(0, 1, n), \"base_h\": rng.uniform(55, 125, n)})\n",
        "    f[\"mult\"] = np.exp(0.045 * (f[\"age\"] - 13) + 0.9 * (f[\"terrain\"] - .5) + 0.8 * (f[\"load\"] - .5)\n",
        "                       + 0.006 * (f[\"pm_gap\"] - 110) + 0.7 * (f[\"humid\"] - .5))\n",
        "    return f\n",
        "\n",
        "\n",
        "def make_data(seed=1, months=MONTHS):\n",
        "    \"\"\"반환: fleet(차량 속성), ev(정비 이력 1건=1행: 발생 DTC 목록, 실제 고장 부품), log(DTC 발생 로그, 행=코드 1건)\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    fleet = make_fleet()\n",
        "    V = len(fleet)\n",
        "    temp = 12 + 14 * np.sin(2 * np.pi * (np.arange(months) - 3) / 12)            # 월평균 기온(가상)\n",
        "    hours = fleet[\"base_h\"].values[:, None] * (1 + 0.25 * np.sin(2 * np.pi * np.arange(months) / 12 + 1)) * rng.lognormal(0, .12, (V, months))\n",
        "    fleet[\"hours\"] = hours.sum(axis=1)                                           # 차량별 18개월 총 가동시간\n",
        "    fleet.attrs[\"month_h\"] = hours.sum(axis=0)                                   # 월별 전체 가동시간(고장률 분모)\n",
        "    evs = []\n",
        "    for p in PID:\n",
        "        s = np.ones(months)\n",
        "        if p in (\"P03\", \"P04\"): s = np.exp(-0.06 * (temp - 12))                   # 한랭기 전기계통 증가\n",
        "        if p in (\"P06\", \"P16\"): s = np.exp(0.05 * (temp - 12))                    # 혹서기 냉각계통 증가\n",
        "        cnt = rng.poisson(RATE[p] * hours / 1000 * fleet[\"mult\"].values[:, None] * s[None, :])\n",
        "        for v, m in zip(*np.nonzero(cnt)):\n",
        "            for _ in range(cnt[v, m]):\n",
        "                evs.append((fleet[\"veh\"][v], int(m), m * 720 + rng.uniform(0, 720), p, float(temp[m] + rng.normal(0, 3))))\n",
        "    ev = pd.DataFrame(evs, columns=[\"veh\", \"month\", \"t\", \"part\", \"temp\"]).sort_values(\"t\").reset_index(drop=True)\n",
        "    ev[\"ev_id\"] = [f\"W{i:05d}\" for i in range(len(ev))]\n",
        "    rows, codes = [], []\n",
        "    for e in ev.itertuples():\n",
        "        c, r = [], rng.random(3)\n",
        "        if r[0] < 0.6: c.append(SPEC[e.part])\n",
        "        if e.part in SHARED and r[1] < 0.6: c.append(SHARED[e.part])\n",
        "        if r[2] < 0.5: c.append(SYMP[PSYS[e.part]])\n",
        "        if not c: c.append(SHARED.get(e.part, SPEC[e.part]))\n",
        "        for q, pr in CASCADE.get(e.part, []):\n",
        "            if rng.random() < pr: c.append(SPEC[q])\n",
        "        if rng.random() < 0.12: c.append(SPEC[PID[rng.integers(0, 16)]])\n",
        "        c = list(dict.fromkeys(c))\n",
        "        rng.shuffle(c)                                                           # 동시 발생 코드의 기록 순서는 임의\n",
        "        codes.append(c)\n",
        "        for k, cd in enumerate(c):\n",
        "            rows.append((e.ev_id, e.veh, e.t + k * 0.1, cd))\n",
        "    ev[\"codes\"] = codes\n",
        "    log = pd.DataFrame(rows, columns=[\"ev_id\", \"veh\", \"ts\", \"code\"])\n",
        "    return fleet, ev, log\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 공통 분석 조각: 코드 멀티핫 행렬, 차량 맥락 피처, 학습/검증 분할(앞 12개월 학습, 뒤 6개월 검증)\n",
        "# ---------------------------------------------------------------------------\n",
        "CIDX = {c: i for i, c in enumerate(CODES)}\n",
        "SPLIT_M = 12\n",
        "\n",
        "\n",
        "def multi_hot(ev):\n",
        "    X = np.zeros((len(ev), len(CODES)))\n",
        "    for r, cs in enumerate(ev[\"codes\"]):\n",
        "        for c in cs:\n",
        "            X[r, CIDX[c]] = 1\n",
        "    return X\n",
        "\n",
        "\n",
        "def context(ev, fleet):\n",
        "    f = fleet.set_index(\"veh\")\n",
        "    return np.column_stack([ev[\"veh\"].map(f[\"age\"]) / 25, ev[\"veh\"].map(f[\"pm_gap\"]) / 200, ev[\"veh\"].map(f[\"terrain\"]),\n",
        "                            ev[\"veh\"].map(f[\"load\"]), (ev[\"temp\"] - 12) / 15])\n",
        "\n",
        "\n",
        "def feats(ev, fleet):\n",
        "    X = multi_hot(ev)\n",
        "    n = X.sum(axis=1, keepdims=True) / 3\n",
        "    return np.hstack([X, n, context(ev, fleet)])\n",
        "\n",
        "\n",
        "def monthly_counts(ev):\n",
        "    return ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0).reindex(columns=PID, fill_value=0)\n",
        "\n",
        "\n",
        "def build_monthly(ev, fleet, top_n=8):\n",
        "    \"\"\"부품별 월간 고장 건수 예측용 표: 지연값(lag), 이동평균, 추세, 기대 건수(가동시간 x 설계 고장률), 기온 상호작용, 계절 성분\"\"\"\n",
        "    mc = ev.groupby([\"month\", \"part\"]).size().unstack(fill_value=0)\n",
        "    top = ev[\"part\"].value_counts().index[:top_n]\n",
        "    mh, tm = fleet.attrs[\"month_h\"], ev.groupby(\"month\")[\"temp\"].mean().values\n",
        "    rows = []\n",
        "    for p in top:\n",
        "        c = mc[p].values.astype(float)\n",
        "        for m in range(3, MONTHS):\n",
        "            L = np.log1p(c)\n",
        "            rows.append({\"part\": p, \"month\": m, \"y\": c[m], \"lag1\": L[m - 1], \"lag2\": L[m - 2], \"lag3\": L[m - 3], \"roll3\": L[m - 3:m].mean(),\n",
        "                         \"trend\": L[m - 1] - L[m - 2], \"roll3_std\": L[m - 3:m].std(), \"log_expected\": np.log(mh[m] / 1000 * RATE[p]),\n",
        "                         \"rate_lag1\": c[m - 1] / (mh[m - 1] / 1000) / RATE[p], \"temp\": tm[m] / 10, \"cold_elec\": max(0, 12 - tm[m]) / 10 * (p in (\"P03\", \"P04\")),\n",
        "                         \"heat_cool\": max(0, tm[m] - 12) / 10 * (p in (\"P06\", \"P16\")), \"sin\": np.sin(2 * np.pi * m / 12), \"cos\": np.cos(2 * np.pi * m / 12)})\n",
        "    D = pd.DataFrame(rows)\n",
        "    return D, [c for c in D.columns if c not in (\"part\", \"month\", \"y\")], top\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 부품별 정비 조치 가이드라인(요약 3단계). 실제 정비 교범이 아니라 교육용으로 만든 예시 문구다.\n",
        "# ---------------------------------------------------------------------------\n",
        "GUIDE = {\n",
        "    \"P01\": \"연료 압력 측정 -> 분사기 누설·막힘 확인 -> 불량 시 분사기 교체\", \"P02\": \"흡기 압력·이물 점검 -> 축 유격 확인 -> 불량 시 터보차저 교체\",\n",
        "    \"P03\": \"단자 전압·부식 점검 -> 충전 계통 확인 -> 불량 시 축전지 교체\", \"P04\": \"발전 전압 측정 -> 벨트·배선 점검 -> 불량 시 발전기 교체\",\n",
        "    \"P05\": \"제어모듈 소프트웨어·커넥터 점검 -> 변속 이력 확인 -> 불량 시 모듈 교체\", \"P06\": \"냉각수량·누수 점검 -> 펌프 작동음 확인 -> 불량 시 펌프 교체\",\n",
        "    \"P07\": \"유압 오일량·압력 점검 -> 펌프 누설 확인 -> 불량 시 펌프 교체\", \"P08\": \"센서 배선·간극 점검 -> 신호 파형 확인 -> 불량 시 센서 교체\",\n",
        "    \"P09\": \"커넥터 접촉·부식 점검 -> 핀 상태 확인 -> 불량 시 커넥터 교체\", \"P10\": \"필터 오염도 점검 -> 연료 흐름 확인 -> 불량 시 필터 교체\",\n",
        "    \"P11\": \"제동 압력 점검 -> 밸브 작동 확인 -> 불량 시 밸브 교체(안전 확인 필수)\", \"P12\": \"조향 반응 점검 -> 액추에이터 작동 확인 -> 불량 시 교체(안전 확인 필수)\",\n",
        "    \"P13\": \"필터 차압 점검 -> 오염 확인 -> 불량 시 필터 교체\", \"P14\": \"통신 링크 상태 점검 -> 모듈 전원 확인 -> 불량 시 모듈 교체\",\n",
        "    \"P15\": \"압력 센서 신호 점검 -> 윤활유량 확인 -> 불량 시 센서 교체\", \"P16\": \"팬 모터 전류 점검 -> 회전 확인 -> 불량 시 모터 교체\"}\n",
        "\n",
        "CONFIG = {\"conf_review\": 0.8, \"retrain_tolerance\": 0.01, \"mape_target\": 0.25}\n",
        "OPS_GUIDE = {                                                              # docs/mnt03_operations_guide.md 요약\n",
        "    \"1. 목적과 범위\": \"DTC 기반 고장 부품 진단 보조와 다빈도 고장품목 식별. 최종 정비 판단은 숙련 정비관.\",\n",
        "    \"2. 입력 데이터\": \"DTC 로그(차량 ID, 시각, 코드), 정비 이력(정비 건 ID, 고장 부품), 차량 속성. 실제 부대 자료 외부 AI 입력 금지.\",\n",
        "    \"3. 일일 운용 절차\": \"로그 검증 -> 정비 건 묶음 -> 진단·신뢰도 산출 -> 낮은 신뢰도 건 승인 대기 -> 조치 후 정비 이력 입력.\",\n",
        "    \"4. 승인(HITL) 기준\": \"신뢰도 0.8 미만·다계통 동시 발생은 숙련 정비관 결정 전 확정 금지, 무응답 시 재확인 표시, 결정 사유 기록.\",\n",
        "    \"5. 재학습 절차\": \"월 1회 신규 정비 이력으로 도전자 학습, 검증 정확도가 나빠지지 않을 때만 교체.\",\n",
        "    \"6. 장애 대응\": \"모델 오류 시 단일 코드 매핑표로 대체, 손상 로그는 수동 확인 목록으로 분리.\",\n",
        "    \"7. 연동 인터페이스\": \"국방 군수 정보체계와 DTC·정비 이력 표준 필드(장비코드·분류군·시각)를 CSV/API로 교환하는 안을 차기 과제로 둔다.\",\n",
        "}\n",
        "\n",
        "fleet, ev, log = make_data()\n",
        "OLD, NEW, OPS = ev[ev[\"month\"] < 12], ev[(ev[\"month\"] >= 12) & (ev[\"month\"] < 15)], ev[ev[\"month\"] >= 15].reset_index(drop=True)\n",
        "t_start = time.perf_counter()\n",
        "\n",
        "\n",
        "def fit(d):\n",
        "    return LogisticRegression(C=3, max_iter=500).fit(feats(d, fleet), d[\"part\"])\n",
        "\n",
        "\n",
        "def retrain(champion, old, new):\n",
        "    \"\"\"도전자를 (기존 + 신규 앞 2/3 기간)으로 학습하고, 신규 뒤 1/3 기간에서 챔피언과 정확도를 비교해 비슷하거나 낫다면 교체.\"\"\"\n",
        "    cut = new[\"month\"].min() + 2\n",
        "    hold = new[new[\"month\"] >= cut]\n",
        "    challenger = fit(pd.concat([old, new[new[\"month\"] < cut]], ignore_index=True))\n",
        "    a_old = (champion.predict(feats(hold, fleet)) == hold[\"part\"].values).mean()\n",
        "    a_new = (challenger.predict(feats(hold, fleet)) == hold[\"part\"].values).mean()\n",
        "    swap = a_new >= a_old - CONFIG[\"retrain_tolerance\"]\n",
        "    return (challenger if swap else champion), {\"챔피언\": round(float(a_old), 3), \"도전자\": round(float(a_new), 3), \"교체\": bool(swap)}\n",
        "\n",
        "\n",
        "def validate(df):\n",
        "    ok = df[\"codes\"].map(lambda cs: len(cs) > 0 and all(c in CODES for c in cs)) & df[\"veh\"].isin(fleet[\"veh\"])\n",
        "    return df[ok].copy(), df[~ok].copy()\n",
        "\n",
        "\n",
        "def predict_safe(model, d):\n",
        "    \"\"\"모델 오류 시 단일 코드 -> 최다 부품 매핑표로 대체한다.\"\"\"\n",
        "    try:\n",
        "        P = model.predict_proba(feats(d, fleet))\n",
        "        return model.classes_[P.argmax(axis=1)], P.max(axis=1), \"모델\"\n",
        "    except Exception:\n",
        "        mp = OLD.assign(first=OLD[\"codes\"].str[0]).groupby(\"first\")[\"part\"].agg(lambda s: s.value_counts().index[0])\n",
        "        return d[\"codes\"].str[0].map(mp).fillna(\"P01\").values, np.full(len(d), 0.5), \"매핑표 대체\"\n",
        "\n",
        "\n",
        "# ---------------- 1. 통합 실행 ----------------\n",
        "champion = fit(OLD)\n",
        "model, rt = retrain(champion, OLD, NEW)\n",
        "print(f\"[교육용가상자료] 재학습: {rt} -> {'도전자로 교체' if rt['교체'] else '챔피언 유지'}\")\n",
        "ops_all = OPS.copy()\n",
        "ops_all.at[3, \"codes\"] = [\"ZZ9\"]; ops_all.at[40, \"codes\"] = []             # 손상 로그 2건 주입\n",
        "ops, manual = validate(ops_all)\n",
        "ops = ops.reset_index(drop=True)\n",
        "pred, conf, src = predict_safe(model, ops)\n",
        "queue = ops[conf < CONFIG[\"conf_review\"]]\n",
        "alerts = [(i, p) for i, p in enumerate(pred) if PSYS[p] in CRIT_SYS and conf[i] >= CONFIG[\"conf_review\"]]\n",
        "acc = (pred == ops[\"part\"].values).mean()\n",
        "hi = conf >= CONFIG[\"conf_review\"]\n",
        "print(f\"운용 정비 건 {len(ops_all)}건: 진단 {len(ops)}건, 수동 확인 {len(manual)}건, 출처 '{src}', 정확도 {acc:.1%} (신뢰도 0.8 이상 {hi.sum()}건은 {(pred[hi] == ops['part'].values[hi]).mean():.1%})\")\n",
        "\n",
        "# 다빈도 품목 월간 소요 예측(마지막 달): 과거 월로 학습한 릿지로 다음 달 건수 예측\n",
        "D, FE, top = build_monthly(ev, fleet)\n",
        "a, b = D[D[\"month\"] < MONTHS - 1], D[D[\"month\"] == MONTHS - 1]\n",
        "fc = np.exp(make_pipeline(StandardScaler(), Ridge(alpha=3.0)).fit(a[FE], np.log(a[\"y\"])).predict(b[FE]))\n",
        "fmape = float(np.mean(np.abs(b[\"y\"].values - fc) / b[\"y\"].values))\n",
        "print(\"다빈도 품목 다음 달 소요 예측: \" + \", \".join(f\"{PNAME[p]} {f:.0f}건(실제 {y:.0f})\" for p, f, y in zip(b[\"part\"].values[:3], fc[:3], b[\"y\"].values[:3])) + f\" / 8개 품목 MAPE {fmape:.1%}\")\n",
        "sec = time.perf_counter() - t_start\n",
        "\n",
        "# ---------------- 2. 자체 점검 ----------------\n",
        "_, _, bad_src = predict_safe(None, ops.head(3))\n",
        "checks = {\n",
        "    \"로그 유실 0건\": len(ops) + len(manual) == len(ops_all),\n",
        "    \"손상 로그 수동 확인 분리\": len(manual) == 2,\n",
        "    \"재학습 파이프라인 실행\": \"교체\" in rt,\n",
        "    \"신뢰도 0~1 범위\": bool(((conf >= 0) & (conf <= 1)).all()),\n",
        "    \"모델 장애 시 매핑표 대체\": bad_src == \"매핑표 대체\",\n",
        "    \"모든 진단에 조치 가이드 연결\": all(p in GUIDE for p in pred),\n",
        "    \"저신뢰 건 자동 확정 금지(대기)\": len(queue) == int((conf < CONFIG[\"conf_review\"]).sum()),\n",
        "    f\"소요 예측 MAPE {CONFIG['mape_target']:.0%} 이하\": fmape <= CONFIG[\"mape_target\"],\n",
        "}\n",
        "print(f\"\\n자체 점검 {sum(checks.values())}/{len(checks)} 통과, 자동 재학습 파이프라인 {'정상 실행' if checks['재학습 파이프라인 실행'] else '실패'}, 전체 실행 {sec:.1f}초\")\n",
        "for k, v in checks.items():\n",
        "    if not v:\n",
        "        print(\" 실패:\", k)\n",
        "doc = sum(bool(v.strip()) for v in OPS_GUIDE.values()) / len(OPS_GUIDE)\n",
        "print(f\"운영 문서 {len(OPS_GUIDE)}개 장 작성률 {doc:.0%} (목표 100%: {'달성' if doc == 1 else '미달'}) - \" + \", \".join(k.split(\". \")[1] for k in OPS_GUIDE))\n",
        "\n",
        "# ---------------- 3. 화면과 브리핑 ----------------\n",
        "fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
        "cnt = pd.Series(pred).value_counts().head(8)\n",
        "ax[0].bar(cnt.index, cnt.values); ax[0].set_title(\"Diagnosed parts (operating period)\")\n",
        "ax[1].hist(conf, bins=20); ax[1].axvline(CONFIG[\"conf_review\"], ls=\"--\", color=\"k\"); ax[1].set_title(\"Confidence (line = review threshold)\")\n",
        "plt.tight_layout(); plt.show()\n",
        "\n",
        "fr = (ev[\"part\"].value_counts() / fleet[\"hours\"].sum() * 1000)\n",
        "cum = ev[\"part\"].value_counts(normalize=True).cumsum()\n",
        "a_cnt = int((cum <= 0.8).sum()) + 1\n",
        "print(\"\\n[정비 담당 장교 브리핑 요약] [교육용가상자료]\")\n",
        "print(f\"1. 18개월 정비 {len(ev)}건: 고장률 상위 \" + \", \".join(f\"{PNAME[p]} {v:.1f}건/1000h\" for p, v in fr.sort_values(ascending=False).head(3).items()))\n",
        "print(f\"2. 상위 {a_cnt}개 품목이 전체 고장의 80% 이상: 예비 수리부속 우선 확보 대상\")\n",
        "print(f\"3. 운용 기간 진단 {len(ops)}건 중 신뢰도 0.8 이상 {int(hi.sum())}건(정확도 {(pred[hi] == ops['part'].values[hi]).mean():.1%}), 정비관 승인 대기 {len(queue)}건(자동 확정 0건)\")\n",
        "print(f\"4. 기동·제동 계통 경보 {len(alerts)}건, 손상 로그 {len(manual)}건 수동 확인\")\n",
        "print(f\"5. 진단 모델: {'신규 정비 이력으로 교체' if rt['교체'] else '기존 유지'}, 소요 예측 MAPE {fmape:.1%}\")\n",
        "print(\"6. 건의: 다빈도 품목 예방정비 주기 재검토, 국방 군수 정보체계와 DTC·정비 이력 표준 필드 연동 인터페이스 정의\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 재학습: {'챔피언': 0.845, '도전자': 0.847, '교체': True} -> 도전자로 교체\n",
        "운용 정비 건 1554건: 진단 1552건, 수동 확인 2건, 출처 '모델', 정확도 83.2% (신뢰도 0.8 이상 931건은 99.2%)\n",
        "다빈도 품목 다음 달 소요 예측: 연료분사기 133건(실제 147), 터보차저 97건(실제 109), 축전지 44건(실제 35) / 8개 품목 MAPE 22.7%\n",
        "\n",
        "자체 점검 8/8 통과, 자동 재학습 파이프라인 정상 실행, 전체 실행 18.1초\n",
        "운영 문서 7개 장 작성률 100% (목표 100%: 달성) - 목적과 범위, 입력 데이터, 일일 운용 절차, 승인(HITL) 기준, 재학습 절차, 장애 대응, 연동 인터페이스\n",
        "\n",
        "[정비 담당 장교 브리핑 요약] [교육용가상자료]\n",
        "1. 18개월 정비 10681건: 고장률 상위 연료분사기 16.3건/1000h, 터보차저 11.6건/1000h, 축전지 11.6건/1000h\n",
        "2. 상위 5개 품목이 전체 고장의 80% 이상: 예비 수리부속 우선 확보 대상\n",
        "3. 운용 기간 진단 1552건 중 신뢰도 0.8 이상 931건(정확도 99.2%), 정비관 승인 대기 621건(자동 확정 0건)\n",
        "4. 기동·제동 계통 경보 710건, 손상 로그 2건 수동 확인\n",
        "5. 진단 모델: 신규 정비 이력으로 교체, 소요 예측 MAPE 22.7%\n",
        "6. 건의: 다빈도 품목 예방정비 주기 재검토, 국방 군수 정보체계와 DTC·정비 이력 표준 필드 연동 인터페이스 정의\n",
        "```\n",
        "</details>\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}