{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# 🎯 [QLT02] 규격·구매요구·납기 연계 품질예측 모델\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/QLT02/QLT02_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",
        "미국 국방부(DoD) 품질보증 부서는 구매요구서의 규격, 계약 납기, 업체 이력과 수령검사 결과를 연결해 불량 로트를 미리 가려내는 품질예측 방식을 공개 자료로 소개해 왔다. 본 프로젝트는 이 사례를 바탕으로 구매요구-계약-납품-수령검사-사후 확정으로 이어지는 교육용 가상 로트 자료를 코드로 만들어, 규격·제원 정합성, 검사 병목, 하자와의 연결, 이상 신호, 설명 근거, 품질 개선 효과를 차례로 분석한다. 학습자는 데이터셋 정의, 결측 정제와 업체 이력 인코딩, 샘플링 판정 Baseline과 탐색적 분석, 은닉 결함 패턴 식별, 품질 위험 지수와 조기경보, 대시보드와 설명 근거, 검사관 재검사(HITL), 대량 불량 납품 스트레스 시험, 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. 군수품 수입/수령 검사 불량 기록(규격 미달, 치수 오차, 외관 결함) 데이터셋 정의\n",
        "**소요시간**: 약 1.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "군수품 수입/수령 검사 불량 기록(규격 미달, 치수 오차) 데이터셋을 정의한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **군수품 수입·수령 검사 불량 기록(규격 미달, 치수 오차, 외관 결함)을 구매요구-계약-납품-검사 흐름에 맞춰 데이터셋으로 정의한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **업무 프로세스 맵(구매요구, 계약, 납품, 수령검사, 불출·사용, 사후 확정)을 그리고 단계별로 생기는 데이터 정리하고 규격화하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `docs/qlt02_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": [
        "\"\"\"군수품 수입·수령 검사 불량 기록(규격 미달·치수 오차·외관 결함) 데이터셋을 구매요구·계약(납기)·납품·검사 흐름에 맞춰 정의하고, 국방 표준 군수 코드(FSC·NIIN) 형식을 반영한 데이터 딕셔너리와 검증기를 만든다.\"\"\"\n",
        "import json\n",
        "import re\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "\n",
        "# 1) 업무 프로세스 맵: 단계마다 어떤 데이터가 생기는지 정리\n",
        "PROCESS = [(\"구매요구\", \"사용부대·보급부서\", \"품목(NIIN), 규격 등급·개정차수, 요구수량, 긴급 여부\"),\n",
        "           (\"계약\", \"조달 담당\", \"업체, 계약 납기(일), 구매요구 대비 규격 변경 여부\"),\n",
        "           (\"납품\", \"업체\", \"로트 번호, 로트 크기, 실제 납품일(지연일)\"),\n",
        "           (\"수령검사\", \"품질 검사관\", \"표본 수, 불량 수(유형별), 치수 편차 측정값, 합부 판정\"),\n",
        "           (\"불출·사용\", \"보급부서·사용부대\", \"긴급 불출 여부, 사용부대 하자·불만 접수\"),\n",
        "           (\"사후 확정\", \"품질 담당\", \"전수검사·하자 반영 확정 불량률\")]\n",
        "print(\"[업무 프로세스 맵] \" + \" -> \".join(p[0] for p in PROCESS))\n",
        "for i, (step, who, data) in enumerate(PROCESS, 1):\n",
        "    print(f\" {i}. {step:5s} | {who:10s} | 생성 데이터: {data}\")\n",
        "\n",
        "# 2) 데이터 딕셔너리(스키마): 단계 / 필드 / 자료형 / 규칙 / 설명\n",
        "S = [\n",
        "    (\"구매요구\", \"niin\", str, {\"pattern\": r\"^\\d{4}-\\d{2}-\\d{3}-\\d{4}$\"}, \"FSC 4자리 + NIIN 9자리(가상) 재고번호\"),\n",
        "    (\"구매요구\", \"fsc\", str, {\"enum\": [\"5305\", \"5330\", \"5935\", \"6135\", \"2530\", \"8415\"]}, \"연방 보급 분류(품목군)\"),\n",
        "    (\"구매요구\", \"spec_class\", str, {\"enum\": [\"precision\", \"general\"]}, \"규격 등급(정밀/일반)\"),\n",
        "    (\"구매요구\", \"tolerance_mm\", float, {\"min\": 0.01, \"max\": 1.0}, \"치수 공차(mm)\"),\n",
        "    (\"구매요구\", \"spec_rev\", int, {\"min\": 1, \"max\": 20}, \"규격서 개정차수\"),\n",
        "    (\"구매요구\", \"pr_qty\", int, {\"min\": 1, \"max\": 100000}, \"구매요구 수량\"),\n",
        "    (\"구매요구\", \"pr_urgent\", int, {\"enum\": [0, 1]}, \"긴급 구매요구 여부\"),\n",
        "    (\"계약\", \"vendor\", str, {\"pattern\": r\"^V\\d{2}$\"}, \"납품업체 코드(가상)\"),\n",
        "    (\"계약\", \"lead_days\", int, {\"min\": 1, \"max\": 365}, \"계약 납기(일)\"),\n",
        "    (\"계약\", \"spec_changed\", int, {\"enum\": [0, 1]}, \"구매요구 후 규격 변경 여부\"),\n",
        "    (\"계약\", \"spec_mismatch\", int, {\"enum\": [0, 1]}, \"구매요구·계약 규격 불일치 여부\"),\n",
        "    (\"납품\", \"lot_id\", str, {\"pattern\": r\"^LOT\\d{2}-\\d{5}$\"}, \"로트 번호\"),\n",
        "    (\"납품\", \"lot_size\", int, {\"min\": 1, \"max\": 100000}, \"로트 크기(개)\"),\n",
        "    (\"납품\", \"delay_days\", int, {\"min\": -30, \"max\": 180}, \"납기 지연일(음수=조기 납품)\"),\n",
        "    (\"수령검사\", \"receipt_date\", str, {\"pattern\": r\"^\\d{4}-\\d{2}-\\d{2}$\"}, \"수령검사일\"),\n",
        "    (\"수령검사\", \"inspector\", str, {\"pattern\": r\"^INS-\\d$\"}, \"검사관 코드\"),\n",
        "    (\"수령검사\", \"sample_n\", int, {\"min\": 1, \"max\": 1250}, \"표본 수\"),\n",
        "    (\"수령검사\", \"defects\", int, {\"min\": 0}, \"표본 불량 수\"),\n",
        "    (\"수령검사\", \"def_spec\", int, {\"min\": 0}, \"규격 미달 수\"),\n",
        "    (\"수령검사\", \"def_dim\", int, {\"min\": 0}, \"치수 오차 수\"),\n",
        "    (\"수령검사\", \"def_visual\", int, {\"min\": 0}, \"외관 결함 수\"),\n",
        "    (\"수령검사\", \"dim_dev_mm\", float, {\"min\": 0, \"max\": 5}, \"치수 편차 평균(mm)\"),\n",
        "    (\"불출·사용\", \"emergency_release\", int, {\"enum\": [0, 1]}, \"긴급 불출(검사 전 선불출) 예외 여부\"),\n",
        "    (\"불출·사용\", \"field_claims\", int, {\"min\": 0}, \"사용부대 하자·불만 접수 건수\"),\n",
        "    (\"사후 확정\", \"confirmed_rate\", float, {\"min\": 0, \"max\": 1}, \"확정 불량률(분석 목표 변수)\"),\n",
        "]\n",
        "SCHEMA = {f: (t, r) for _, f, t, r, _ in S}\n",
        "CROSS_RULES = [\"표본 수 <= 로트 크기\", \"유형별 불량 합 = 표본 불량 수\", \"불량 수 <= 표본 수\", \"niin 앞 4자리 = fsc\"]\n",
        "\n",
        "\n",
        "def validate(rec):\n",
        "    errs = [f\"{f}: 필수 필드 누락\" for f in SCHEMA if f not in rec]\n",
        "    for f, (t, r) in SCHEMA.items():\n",
        "        if f not in rec:\n",
        "            continue\n",
        "        v = rec[f]\n",
        "        if not isinstance(v, t) or isinstance(v, bool):\n",
        "            errs.append(f\"{f}: 자료형 오류({type(v).__name__})\"); continue\n",
        "        if \"enum\" in r and v not in r[\"enum\"]:\n",
        "            errs.append(f\"{f}: 허용되지 않은 값 '{v}'\")\n",
        "        if \"min\" in r and v < r[\"min\"]:\n",
        "            errs.append(f\"{f}: 최소값 {r['min']} 미만\")\n",
        "        if \"max\" in r and v > r[\"max\"]:\n",
        "            errs.append(f\"{f}: 최대값 {r['max']} 초과\")\n",
        "        if \"pattern\" in r and not re.match(r[\"pattern\"], v):\n",
        "            errs.append(f\"{f}: 형식 불일치 '{v}'\")\n",
        "    if not errs:\n",
        "        if rec[\"sample_n\"] > rec[\"lot_size\"]:\n",
        "            errs.append(\"교차검증: 표본 수가 로트 크기보다 큼\")\n",
        "        if rec[\"def_spec\"] + rec[\"def_dim\"] + rec[\"def_visual\"] != rec[\"defects\"]:\n",
        "            errs.append(\"교차검증: 유형별 불량 합 불일치\")\n",
        "        if rec[\"defects\"] > rec[\"sample_n\"]:\n",
        "            errs.append(\"교차검증: 불량 수가 표본 수보다 큼\")\n",
        "        if rec[\"niin\"][:4] != rec[\"fsc\"]:\n",
        "            errs.append(\"교차검증: 재고번호와 품목군 불일치\")\n",
        "    return errs\n",
        "\n",
        "\n",
        "# 3) [교육용가상자료] 샘플 20건 생성\n",
        "rng = np.random.default_rng(3)\n",
        "recs = []\n",
        "for i in range(20):\n",
        "    fsc = str(rng.choice([\"5305\", \"5330\", \"5935\", \"6135\", \"2530\", \"8415\"]))\n",
        "    n = int(rng.choice([50, 80, 125])); dfx = [int(x) for x in rng.multinomial(int(rng.integers(0, 6)), [0.4, 0.4, 0.2])]\n",
        "    prec = bool(rng.random() < 0.4)\n",
        "    recs.append({\"niin\": f\"{fsc}-01-{int(rng.integers(100, 999))}-{int(rng.integers(1000, 9999))}\", \"fsc\": fsc,\n",
        "                 \"spec_class\": \"precision\" if prec else \"general\", \"tolerance_mm\": round(float(rng.uniform(0.05, 0.1) if prec else rng.uniform(0.2, 0.5)), 3),\n",
        "                 \"spec_rev\": int(rng.integers(1, 6)), \"pr_qty\": int(rng.integers(500, 5000)), \"pr_urgent\": int(rng.random() < 0.2),\n",
        "                 \"vendor\": f\"V{int(rng.integers(1, 13)):02d}\", \"lead_days\": int(rng.integers(30, 150)), \"spec_changed\": int(rng.random() < 0.1),\n",
        "                 \"spec_mismatch\": 0, \"lot_id\": f\"LOT01-{i:05d}\", \"lot_size\": int(rng.integers(300, 3000)), \"delay_days\": int(rng.integers(-3, 20)),\n",
        "                 \"receipt_date\": f\"2025-{int(rng.integers(1, 13)):02d}-{int(rng.integers(1, 29)):02d}\", \"inspector\": f\"INS-{int(rng.integers(1, 5))}\",\n",
        "                 \"sample_n\": n, \"defects\": sum(dfx), \"def_spec\": dfx[0], \"def_dim\": dfx[1], \"def_visual\": dfx[2],\n",
        "                 \"dim_dev_mm\": round(float(rng.uniform(0.01, 0.2)), 4), \"emergency_release\": 0, \"field_claims\": int(rng.poisson(1)),\n",
        "                 \"confirmed_rate\": round(float(rng.uniform(0.005, 0.08)), 4)})\n",
        "ok = sum(not validate(r) for r in recs)\n",
        "print(f\"\\n정상 샘플 {len(recs)}건 중 검증 통과 {ok}건, 오류 {len(recs) - ok}건\")\n",
        "\n",
        "# 4) 오류 주입 시험\n",
        "bad = json.loads(json.dumps(recs[0]))\n",
        "bad[\"niin\"] = \"5305011234567\"          # 하이픈 없는 재고번호\n",
        "bad[\"vendor\"] = \"업체A\"                  # 코드 형식 위반\n",
        "bad[\"spec_class\"] = \"high\"             # 허용값 위반\n",
        "bad[\"lot_size\"] = -10                  # 범위 위반\n",
        "del bad[\"receipt_date\"]                # 필수 필드 누락\n",
        "print(\"\\n[오류 주입 결과]\")\n",
        "for e in validate(bad):\n",
        "    print(\" -\", e)\n",
        "bad2 = json.loads(json.dumps(recs[1])); bad2[\"def_dim\"] += 2; bad2[\"niin\"] = \"9999\" + bad2[\"niin\"][4:]\n",
        "print(\" - (교차검증) \" + \" / \".join(validate(bad2)))\n",
        "\n",
        "# 5) 데이터 딕셔너리 요약과 목표 점검\n",
        "dd = pd.DataFrame([(s, f, t.__name__, json.dumps(r, ensure_ascii=False)) for s, f, t, r, _ in S], columns=[\"단계\", \"필드\", \"자료형\", \"규칙\"])\n",
        "print(\"\\n[데이터 딕셔너리 요약: 단계별 필드 수]\")\n",
        "print(dd.groupby(\"단계\", sort=False)[\"필드\"].agg([\"count\", lambda x: \", \".join(x)]).rename(columns={\"count\": \"필드 수\", \"<lambda_0>\": \"필드\"}).to_string())\n",
        "print(f\"\\n피처(필드) {len(S)}개 정의 (목표 20개 이상: {'달성' if len(S) >= 20 else '미달'}), 교차검증 규칙 {len(CROSS_RULES)}종, 프로세스 맵 {len(PROCESS)}단계 작성\")\n",
        "print(\"분석 목표: 구매요구·규격·납기 정보로 로트 확정 불량률을 수령검사 시점에 미리 예측해 고위험 로트를 먼저 정밀 검사한다.\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[업무 프로세스 맵] 구매요구 -> 계약 -> 납품 -> 수령검사 -> 불출·사용 -> 사후 확정\n",
        " 1. 구매요구  | 사용부대·보급부서  | 생성 데이터: 품목(NIIN), 규격 등급·개정차수, 요구수량, 긴급 여부\n",
        " 2. 계약    | 조달 담당      | 생성 데이터: 업체, 계약 납기(일), 구매요구 대비 규격 변경 여부\n",
        " 3. 납품    | 업체         | 생성 데이터: 로트 번호, 로트 크기, 실제 납품일(지연일)\n",
        " 4. 수령검사  | 품질 검사관     | 생성 데이터: 표본 수, 불량 수(유형별), 치수 편차 측정값, 합부 판정\n",
        " 5. 불출·사용 | 보급부서·사용부대  | 생성 데이터: 긴급 불출 여부, 사용부대 하자·불만 접수\n",
        " 6. 사후 확정 | 품질 담당      | 생성 데이터: 전수검사·하자 반영 확정 불량률\n",
        "\n",
        "정상 샘플 20건 중 검증 통과 20건, 오류 0건\n",
        "\n",
        "[오류 주입 결과]\n",
        " - receipt_date: 필수 필드 누락\n",
        " - niin: 형식 불일치 '5305011234567'\n",
        " - spec_class: 허용되지 않은 값 'high'\n",
        " - vendor: 형식 불일치 '업체A'\n",
        " - lot_size: 최소값 1 미만\n",
        " - (교차검증) 교차검증: 유형별 불량 합 불일치 / 교차검증: 재고번호와 품목군 불일치\n",
        "\n",
        "[데이터 딕셔너리 요약: 단계별 필드 수]\n",
        "       필드 수                                                                                     필드\n",
        "단계                                                                                                \n",
        "구매요구      7                       niin, fsc, spec_class, tolerance_mm, spec_rev, pr_qty, pr_urgent\n",
        "계약        4                                         vendor, lead_days, spec_changed, spec_mismatch\n",
        "납품        3                                                           lot_id, lot_size, delay_days\n",
        "수령검사      8  receipt_date, inspector, sample_n, defects, def_spec, def_dim, def_visual, dim_dev_mm\n",
        "불출·사용     2                                                        emergency_release, field_claims\n",
        "사후 확정     1                                                                         confirmed_rate\n",
        "\n",
        "피처(필드) 25개 정의 (목표 20개 이상: 달성), 교차검증 규칙 4종, 프로세스 맵 6단계 작성\n",
        "분석 목표: 구매요구·규격·납기 정보로 로트 확정 불량률을 수령검사 시점에 미리 예측해 고위험 로트를 먼저 정밀 검사한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 02. 검사 측정치 결측치 정제, 납품업체별 불량 이력 인코딩 전처리 모듈 구현\n",
        "**소요시간**: 약 2.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "검사 측정치 결측치 정제 및 업체별 불량 이력 인코딩 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **검사 측정치 결측을 정제하고 납품업체별 불량 이력을 정보 누수 없이 인코딩하는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **중복 로트, 업체 코드 표기, 날짜 형식, 재고번호 형식, 범위 오류 등 정제 규칙 10종을 적용하고 치수 측정 결측은 KNN으로 보간 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/etl/qlt02_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": [
        "\"\"\"수령검사 원시 기록의 결측·오기·중복을 정제 규칙 10종으로 바로잡고(측정치 결측은 KNN 보간), 납품업체별 과거 불량 이력을 정보 누수 없이 인코딩하는 전처리 모듈을 구현한다. 긴급 불출 등 군수 특수 예외 기록은 삭제하지 않고 보존한다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "from sklearn.impute import KNNImputer\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "\n",
        "def make_raw(clean, seed=5):\n",
        "    \"\"\"[교육용가상자료] 현장 입력 오류를 흉내 낸 원시 데이터: 결측·형식 오류·단위 혼용·중복 등을 일부러 넣는다.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    r = clean.copy()\n",
        "    r[\"receipt_date\"] = r[\"receipt_date\"].dt.strftime(\"%Y-%m-%d\").astype(object)\n",
        "    idx = lambda k: rng.choice(len(r), k, replace=False)\n",
        "    r.loc[idx(40), \"dim_dev_mm\"] = np.nan                                      # 측정치 결측\n",
        "    i = idx(15); r.loc[i, \"dim_dev_mm\"] = r.loc[i, \"dim_dev_mm\"] * 1000          # mm/um 단위 혼용\n",
        "    i = idx(20); r.loc[i, \"vendor\"] = [\" \" + v.lower() + \" \" for v in r.loc[i, \"vendor\"]]   # 대소문자·공백\n",
        "    i = idx(15); r.loc[i, \"receipt_date\"] = [d.replace(\"-\", \".\") for d in r.loc[i, \"receipt_date\"]]   # 날짜 형식\n",
        "    i = idx(10); r.loc[i, \"niin\"] = [n.replace(\"-\", \"\") for n in r.loc[i, \"niin\"]]   # 재고번호 하이픈 누락\n",
        "    i = idx(8); r.loc[i, \"lot_size\"] = -r.loc[i, \"lot_size\"]                     # 부호 오류\n",
        "    i = idx(6); r.loc[i, \"defects\"] = r.loc[i, \"defects\"] + 3                    # 유형별 합 불일치\n",
        "    i = idx(5); r.loc[i, \"spec_class\"] = \"PRECISION\"                             # 코드 표기 불일치\n",
        "    i = idx(5); r.loc[i, \"inspector\"] = np.nan                                  # 검사관 누락\n",
        "    r = pd.concat([r, r.iloc[idx(12)]], ignore_index=True)                       # 중복 입력\n",
        "    return r\n",
        "\n",
        "\n",
        "def clean(raw):\n",
        "    d, log = raw.copy(), []\n",
        "\n",
        "    def rule(name, mask, fix):\n",
        "        n = int(mask.sum())\n",
        "        if n:\n",
        "            fix(mask)\n",
        "        log.append((name, n))\n",
        "\n",
        "    rule(\"R01 중복 로트 제거\", d.duplicated(\"lot_id\"), lambda m: None)\n",
        "    d = d[~d.duplicated(\"lot_id\")].reset_index(drop=True)\n",
        "    rule(\"R02 업체 코드 공백·대소문자 정규화\", d[\"vendor\"] != d[\"vendor\"].str.strip().str.upper(),\n",
        "         lambda m: d.__setitem__(\"vendor\", d[\"vendor\"].str.strip().str.upper()))\n",
        "    rule(\"R03 날짜 형식 통일(YYYY-MM-DD)\", d[\"receipt_date\"].str.contains(r\"\\.\"),\n",
        "         lambda m: d.__setitem__(\"receipt_date\", d[\"receipt_date\"].str.replace(\".\", \"-\", regex=False)))\n",
        "    d[\"receipt_date\"] = pd.to_datetime(d[\"receipt_date\"])\n",
        "    bad_niin = ~d[\"niin\"].str.match(r\"^\\d{4}-\\d{2}-\\d{3}-\\d{4}$\")\n",
        "    fixed_niin = d[\"niin\"].str.replace(r\"^(\\d{4})(\\d{2})(\\d{3})(\\d{4})$\", r\"\\1-\\2-\\3-\\4\", regex=True)\n",
        "    rule(\"R04 재고번호(NIIN) 형식 복원\", bad_niin, lambda m: d.__setitem__(\"niin\", fixed_niin))\n",
        "    rule(\"R05 규격 등급 코드 소문자 통일\", d[\"spec_class\"] != d[\"spec_class\"].str.lower(),\n",
        "         lambda m: d.__setitem__(\"spec_class\", d[\"spec_class\"].str.lower()))\n",
        "    rule(\"R06 로트 크기 부호 오류 수정\", d[\"lot_size\"] < 0, lambda m: d.__setitem__(\"lot_size\", d[\"lot_size\"].abs()))\n",
        "    rule(\"R07 측정 단위 환산(um -> mm)\", d[\"dim_dev_mm\"] > 5, lambda m: d.__setitem__(\"dim_dev_mm\", d[\"dim_dev_mm\"].where(~m, d[\"dim_dev_mm\"] / 1000)))\n",
        "    rule(\"R08 유형별 합에 맞춰 불량 수 재계산\", d[\"defects\"] != d[[\"def_spec\", \"def_dim\", \"def_visual\"]].sum(axis=1),\n",
        "         lambda m: d.__setitem__(\"defects\", d[[\"def_spec\", \"def_dim\", \"def_visual\"]].sum(axis=1)))\n",
        "    rule(\"R09 검사관 누락 -> 'UNKNOWN' 표시(삭제하지 않음)\", d[\"inspector\"].isna(), lambda m: d.__setitem__(\"inspector\", d[\"inspector\"].fillna(\"UNKNOWN\")))\n",
        "    miss = d[\"dim_dev_mm\"].isna()\n",
        "    num = [\"dim_dev_mm\", \"tolerance_mm\", \"defects\", \"sample_n\", \"lot_size\"]\n",
        "    Z = d[num].astype(float); mu, sd = Z.mean(), Z.std()\n",
        "    filled = KNNImputer(n_neighbors=5).fit_transform((Z - mu) / sd) * sd.values + mu.values\n",
        "    rule(\"R10 치수 편차 결측 KNN 보간(k=5)\", miss, lambda m: d.__setitem__(\"dim_dev_mm\", np.where(m, filled[:, 0].round(4), d[\"dim_dev_mm\"])))\n",
        "    d[\"dim_imputed\"] = miss.astype(int)                                         # 보간 여부 표시(추적성)\n",
        "    return d, log\n",
        "\n",
        "\n",
        "def integrity(d):\n",
        "    \"\"\"정합성 검증 항목: 모두 True여야 통과.\"\"\"\n",
        "    return {\"로트 번호 유일\": d[\"lot_id\"].is_unique,\n",
        "            \"업체 코드 형식\": d[\"vendor\"].str.match(r\"^V\\d{2}$\").all(),\n",
        "            \"재고번호 형식\": d[\"niin\"].str.match(r\"^\\d{4}-\\d{2}-\\d{3}-\\d{4}$\").all(),\n",
        "            \"재고번호-품목군 일치\": (d[\"niin\"].str[:4] == d[\"fsc\"]).all(),\n",
        "            \"로트 크기 양수\": (d[\"lot_size\"] > 0).all(),\n",
        "            \"표본 <= 로트\": (d[\"sample_n\"] <= d[\"lot_size\"]).all(),\n",
        "            \"불량 유형 합 일치\": (d[\"defects\"] == d[[\"def_spec\", \"def_dim\", \"def_visual\"]].sum(axis=1)).all(),\n",
        "            \"치수 편차 결측 없음·범위\": d[\"dim_dev_mm\"].between(0, 5).all(),\n",
        "            \"날짜형 변환\": pd.api.types.is_datetime64_any_dtype(d[\"receipt_date\"]),\n",
        "            \"규격 등급 허용값\": d[\"spec_class\"].isin([\"precision\", \"general\"]).all()}\n",
        "\n",
        "\n",
        "def encode_vendor_history(d):\n",
        "    \"\"\"업체별 불량 이력 인코딩: 같은 업체의 '이전' 로트 확정 불량률만 써서 평균·건수·최근 5로트 평균을 만든다.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_prev_n\"] = d.groupby(\"vendor\").cumcount()\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean())\n",
        "    prior = d[\"confirmed_rate\"].mean()                                           # 이력이 적은 업체는 전체 평균 쪽으로 수축\n",
        "    w = d[\"vendor_prev_n\"] / (d[\"vendor_prev_n\"] + 10)\n",
        "    d[\"vendor_hist_enc\"] = (w * d[\"vendor_hist_rate\"].fillna(prior) + (1 - w) * prior).round(4)\n",
        "    return d\n",
        "\n",
        "\n",
        "truth = make_lots(1200, seed=42)\n",
        "raw = make_raw(truth)\n",
        "print(f\"원시 기록 {len(raw)}건 (중복 포함), 결측 치수값 {raw['dim_dev_mm'].isna().sum()}건\")\n",
        "pre = integrity(raw)\n",
        "print(f\"정제 전 정합성 검증 {sum(pre.values())}/{len(pre)} 통과, 실패 항목: {', '.join(k for k, v in pre.items() if not v)}\")\n",
        "cleaned, log = clean(raw)\n",
        "print(\"\\n[정제 규칙 적용 내역]\")\n",
        "for name, n in log:\n",
        "    print(f\" {name:34s} {n:4d}건\")\n",
        "chk = integrity(cleaned)\n",
        "rate = np.mean(list(chk.values()))\n",
        "print(f\"\\n정제 규칙 {len(log)}종 적용, 정합성 검증 {sum(chk.values())}/{len(chk)} 통과 ({rate:.0%}) (목표 100%: {'달성' if rate == 1 else '미달'})\")\n",
        "print(f\"긴급 불출 예외 기록 보존: 원본 {truth['emergency_release'].sum()}건 -> 정제 후 {cleaned['emergency_release'].sum()}건 (삭제 없음)\")\n",
        "\n",
        "m = cleaned[\"dim_imputed\"] == 1\n",
        "ref = truth.set_index(\"lot_id\").loc[cleaned.loc[m, \"lot_id\"], \"dim_dev_mm\"].values\n",
        "print(f\"KNN 보간 {m.sum()}건 평균 절대오차 {np.mean(np.abs(cleaned.loc[m, 'dim_dev_mm'].values - ref)):.4f} mm (단순 평균 대치 {np.mean(np.abs(truth['dim_dev_mm'].mean() - ref)):.4f} mm)\")\n",
        "\n",
        "print(\"\\n[전후 기술통계 비교]\")\n",
        "cmp = pd.DataFrame({\"정제 전 평균\": raw[[\"lot_size\", \"dim_dev_mm\", \"defects\"]].mean(), \"정제 후 평균\": cleaned[[\"lot_size\", \"dim_dev_mm\", \"defects\"]].mean(),\n",
        "                    \"원본(정답) 평균\": truth[[\"lot_size\", \"dim_dev_mm\", \"defects\"]].mean()}).round(3)\n",
        "print(cmp.to_string())\n",
        "\n",
        "enc = encode_vendor_history(cleaned)\n",
        "print(\"\\n[업체별 불량 이력 인코딩 예시: 마지막 로트 기준]\")\n",
        "print(enc.groupby(\"vendor\").tail(1)[[\"vendor\", \"vendor_prev_n\", \"vendor_hist_enc\", \"vendor_recent_rate\"]].sort_values(\"vendor_hist_enc\", ascending=False).head(6).round(4).to_string(index=False))\n",
        "print(\"주의: 현재 로트의 확정 불량률을 인코딩에 넣으면 정보 누수가 생기므로 이전 로트만 사용했다.\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "원시 기록 1212건 (중복 포함), 결측 치수값 40건\n",
        "정제 전 정합성 검증 1/10 통과, 실패 항목: 로트 번호 유일, 업체 코드 형식, 재고번호 형식, 로트 크기 양수, 표본 <= 로트, 불량 유형 합 일치, 치수 편차 결측 없음·범위, 날짜형 변환, 규격 등급 허용값\n",
        "\n",
        "[정제 규칙 적용 내역]\n",
        " R01 중복 로트 제거                         12건\n",
        " R02 업체 코드 공백·대소문자 정규화                20건\n",
        " R03 날짜 형식 통일(YYYY-MM-DD)             15건\n",
        " R04 재고번호(NIIN) 형식 복원                 10건\n",
        " R05 규격 등급 코드 소문자 통일                   5건\n",
        " R06 로트 크기 부호 오류 수정                    8건\n",
        " R07 측정 단위 환산(um -> mm)               14건\n",
        " R08 유형별 합에 맞춰 불량 수 재계산                6건\n",
        " R09 검사관 누락 -> 'UNKNOWN' 표시(삭제하지 않음)    5건\n",
        " R10 치수 편차 결측 KNN 보간(k=5)             40건\n",
        "\n",
        "정제 규칙 10종 적용, 정합성 검증 10/10 통과 (100%) (목표 100%: 달성)\n",
        "긴급 불출 예외 기록 보존: 원본 76건 -> 정제 후 76건 (삭제 없음)\n",
        "KNN 보간 40건 평균 절대오차 0.0233 mm (단순 평균 대치 0.0543 mm)\n",
        "\n",
        "[전후 기술통계 비교]\n",
        "             정제 전 평균   정제 후 평균  원본(정답) 평균\n",
        "lot_size    1970.468  1983.859   1983.859\n",
        "dim_dev_mm     0.973     0.094      0.094\n",
        "defects        5.051     5.061      5.061\n",
        "\n",
        "[업체별 불량 이력 인코딩 예시: 마지막 로트 기준]\n",
        "vendor  vendor_prev_n  vendor_hist_enc  vendor_recent_rate\n",
        "   V12             86           0.1072              0.1152\n",
        "   V11             80           0.0699              0.0366\n",
        "   V10             79           0.0692              0.1406\n",
        "   V08             57           0.0487              0.0278\n",
        "   V09             81           0.0450              0.0539\n",
        "   V07             69           0.0402              0.0307\n",
        "주의: 현재 로트의 확정 불량률을 인코딩에 넣으면 정보 누수가 생기므로 이전 로트만 사용했다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 03. 불량률 통계 기반 1차 로트(Lot) 합격/불합격 판정 Baseline 알고리즘 구현\n",
        "**소요시간**: 약 2.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "불량률 통계 기반 로트(Lot) 합격/불합격 판정 Baseline 알고리즘을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **표본 불량 수가 합격판정수 이하이면 합격시키는 통계적 샘플링 기반 로트 합격/불합격 판정 Baseline을 구현하고, 탐색적 분석으로 불량 관련 요인을 찾는다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **로트 크기별 표본 수와 합격판정수로 판정한 뒤 확정 불량률과 비교해 오합격·오불합격을 측정 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `notebooks/qlt02_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": [
        "\"\"\"표본 불량 수가 합격판정수(c) 이하이면 합격시키는 통계적 샘플링 기반 로트 합격/불합격 판정 Baseline을 구현해 오판정(오합격·오불합격)을 측정하고, 탐색적 데이터 분석(차트 8종·상관 검정·파레토 ABC)으로 불량과 관련된 규격·구매요구·납기 요인을 찾는다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy import stats\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "d = make_lots(1200, seed=42)\n",
        "\n",
        "# 1) Baseline: 단일 샘플링 판정 (표본 불량 수 <= c 이면 합격)\n",
        "d[\"base_accept\"] = (d[\"defects\"] <= d[\"accept_c\"]).astype(int)\n",
        "cm = pd.crosstab(d[\"bad_lot\"].map({0: \"실제 양호\", 1: \"실제 불량\"}), d[\"base_accept\"].map({1: \"합격 판정\", 0: \"불합격 판정\"}))\n",
        "print(\"[Baseline 판정 혼동표] (불량 로트 기준: 확정 불량률 5% 초과)\")\n",
        "print(cm.to_string())\n",
        "acc = (d[\"base_accept\"] == 1 - d[\"bad_lot\"]).mean()\n",
        "cons = ((d[\"base_accept\"] == 1) & (d[\"bad_lot\"] == 1)).sum() / d[\"bad_lot\"].sum()\n",
        "prod = ((d[\"base_accept\"] == 0) & (d[\"bad_lot\"] == 0)).sum() / (d[\"bad_lot\"] == 0).sum()\n",
        "print(f\"판정 정확도 {acc:.1%}, 오합격(소비자 위험) {cons:.1%}, 오불합격(생산자 위험) {prod:.1%}\")\n",
        "print(\"한계: 표본 불량 수만 보므로 업체 이력·규격 변경·납기 지연 같은 사전 정보를 쓰지 못하고, 경계선 로트에서 오판정이 몰린다.\")\n",
        "\n",
        "# 2) 상관 분석: 확정 불량률과 각 요인의 스피어만 상관계수·p-value\n",
        "cand = {\"delay_days\": d[\"delay_days\"], \"lead_days\": d[\"lead_days\"], \"precision_spec\": (d[\"spec_class\"] == \"precision\").astype(int),\n",
        "        \"pr_urgent\": d[\"pr_urgent\"], \"spec_changed\": d[\"spec_changed\"], \"spec_mismatch\": d[\"spec_mismatch\"],\n",
        "        \"dim_dev/tol\": d[\"dim_dev_mm\"] / d[\"tolerance_mm\"], \"lot_size\": d[\"lot_size\"], \"spec_rev\": d[\"spec_rev\"], \"unit_cost\": d[\"unit_cost\"]}\n",
        "rows = []\n",
        "for k, v in cand.items():\n",
        "    r, p = stats.spearmanr(v, d[\"confirmed_rate\"])\n",
        "    rows.append({\"요인\": k, \"상관계수\": round(r, 3), \"p-value\": f\"{p:.1e}\", \"유의(p<0.05)\": \"예\" if p < 0.05 else \"아니오\"})\n",
        "cr = pd.DataFrame(rows).sort_values(\"상관계수\", key=abs, ascending=False)\n",
        "print(\"\\n[확정 불량률과 요인별 상관 검정]\")\n",
        "print(cr.to_string(index=False))\n",
        "n_sig = (cr[\"유의(p<0.05)\"] == \"예\").sum()\n",
        "print(f\"유의미한 상관 피처 {n_sig}개 (목표 5개 이상: {'달성' if n_sig >= 5 else '미달'})\")\n",
        "chi = stats.chi2_contingency(pd.crosstab(d[\"vendor\"], d[\"bad_lot\"]))\n",
        "print(f\"업체와 불량 로트 여부의 독립성 카이제곱 검정 p-value {chi.pvalue:.1e} -> 업체별 차이 {'있음' if chi.pvalue < 0.05 else '없음'}\")\n",
        "\n",
        "# 3) 파레토(ABC) 분석: 불량 개수 추정치(로트 크기 x 확정 불량률) 기준 업체 순위\n",
        "d[\"def_units\"] = d[\"lot_size\"] * d[\"confirmed_rate\"]\n",
        "par = d.groupby(\"vendor\")[\"def_units\"].sum().sort_values(ascending=False)\n",
        "cum = par.cumsum() / par.sum()\n",
        "abc = np.where(cum <= 0.5, \"A\", np.where(cum <= 0.8, \"B\", \"C\"))\n",
        "top = par.index[: int(np.ceil(len(par) * 0.2))]\n",
        "print(f\"\\n[파레토 ABC] 상위 20% 업체 {list(top)}가 추정 불량 개수의 {par[top].sum() / par.sum():.0%}를 차지 -> 핵심 관리 대상\")\n",
        "print(\"A등급:\", list(par.index[abc == \"A\"]), \"/ B등급:\", list(par.index[abc == \"B\"]))\n",
        "\n",
        "# 4) EDA 차트 8종\n",
        "fig, ax = plt.subplots(2, 4, figsize=(17, 8))\n",
        "ax[0, 0].hist(d[\"confirmed_rate\"], bins=40, color=\"tab:blue\"); ax[0, 0].axvline(BAD_RATE, c=\"r\", ls=\"--\"); ax[0, 0].set_title(\"1. Confirmed defect rate\")\n",
        "d.groupby(\"month\")[\"confirmed_rate\"].mean().plot(ax=ax[0, 1], marker=\"o\"); ax[0, 1].set_title(\"2. Monthly mean defect rate\")\n",
        "par.plot.bar(ax=ax[0, 2], color=\"tab:orange\"); ax2 = ax[0, 2].twinx(); ax2.plot(range(len(cum)), cum.values, \"k-o\"); ax[0, 2].set_title(\"3. Vendor Pareto\")\n",
        "d.boxplot(column=\"confirmed_rate\", by=\"spec_class\", ax=ax[0, 3]); ax[0, 3].set_title(\"4. Rate by spec class\")\n",
        "ax[1, 0].scatter(d[\"delay_days\"], d[\"confirmed_rate\"], s=6, alpha=0.5); ax[1, 0].set_title(\"5. Delivery delay vs rate\"); ax[1, 0].set_xlabel(\"delay days\")\n",
        "pd.crosstab(d[\"fsc\"], d[\"bad_lot\"], normalize=\"index\")[1].plot.bar(ax=ax[1, 1], color=\"tab:red\"); ax[1, 1].set_title(\"6. Bad-lot share by FSC\")\n",
        "d[[\"def_spec\", \"def_dim\", \"def_visual\"]].sum().plot.pie(ax=ax[1, 2], autopct=\"%.0f%%\", labels=[\"spec\", \"dimension\", \"visual\"]); ax[1, 2].set_ylabel(\"\"); ax[1, 2].set_title(\"7. Defect type mix\")\n",
        "cmat = pd.DataFrame(cand).assign(rate=d[\"confirmed_rate\"]).corr(method=\"spearman\")\n",
        "ax[1, 3].imshow(cmat, cmap=\"coolwarm\", vmin=-1, vmax=1); ax[1, 3].set_xticks(range(len(cmat))); ax[1, 3].set_xticklabels(cmat.columns, rotation=90, fontsize=7)\n",
        "ax[1, 3].set_yticks(range(len(cmat))); ax[1, 3].set_yticklabels(cmat.columns, fontsize=7); ax[1, 3].set_title(\"8. Spearman correlation\")\n",
        "plt.suptitle(\"QLT02 EDA - synthetic receiving inspection data\"); plt.tight_layout(); plt.show()\n",
        "print(f\"\\nEDA 차트 {len(fig.axes) - 1}종 생성 (목표 8종 이상: 달성)\")\n",
        "\n",
        "print(\"\\n[핵심 발견점]\")\n",
        "print(f\" 1. 정밀 규격 로트의 평균 확정 불량률 {d[d.spec_class == 'precision'].confirmed_rate.mean():.1%}, 일반 규격 {d[d.spec_class == 'general'].confirmed_rate.mean():.1%}\")\n",
        "print(f\" 2. 규격 불일치 로트 {d[d.spec_mismatch == 1].confirmed_rate.mean():.1%} vs 일치 {d[d.spec_mismatch == 0].confirmed_rate.mean():.1%}\")\n",
        "print(f\" 3. 납기 14일 이상 지연 로트 {d[d.delay_days >= 14].confirmed_rate.mean():.1%} vs 정시 납품 {d[d.delay_days <= 0].confirmed_rate.mean():.1%}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[Baseline 판정 혼동표] (불량 로트 기준: 확정 불량률 5% 초과)\n",
        "base_accept  불합격 판정  합격 판정\n",
        "bad_lot                   \n",
        "실제 불량           203     79\n",
        "실제 양호            34    884\n",
        "판정 정확도 90.6%, 오합격(소비자 위험) 28.0%, 오불합격(생산자 위험) 3.7%\n",
        "한계: 표본 불량 수만 보므로 업체 이력·규격 변경·납기 지연 같은 사전 정보를 쓰지 못하고, 경계선 로트에서 오판정이 몰린다.\n",
        "\n",
        "[확정 불량률과 요인별 상관 검정]\n",
        "            요인   상관계수 p-value 유의(p<0.05)\n",
        "   dim_dev/tol  0.459 1.1e-63          예\n",
        "precision_spec  0.442 1.7e-58          예\n",
        "    delay_days  0.288 2.5e-24          예\n",
        "     pr_urgent  0.277 1.6e-22          예\n",
        " spec_mismatch  0.267 5.3e-21          예\n",
        "  spec_changed  0.211 1.4e-13          예\n",
        "     lead_days -0.182 2.1e-10          예\n",
        "      lot_size  0.024 4.1e-01        아니오\n",
        "     unit_cost  0.014 6.3e-01        아니오\n",
        "      spec_rev  0.002 9.3e-01        아니오\n",
        "유의미한 상관 피처 7개 (목표 5개 이상: 달성)\n",
        "업체와 불량 로트 여부의 독립성 카이제곱 검정 p-value 2.3e-55 -> 업체별 차이 있음\n",
        "\n",
        "[파레토 ABC] 상위 20% 업체 ['V12', 'V11', 'V10']가 추정 불량 개수의 43%를 차지 -> 핵심 관리 대상\n",
        "A등급: ['V12', 'V11', 'V10'] / B등급: ['V05', 'V09', 'V03', 'V06', 'V04']\n",
        "\n",
        "EDA 차트 8종 생성 (목표 8종 이상: 달성)\n",
        "\n",
        "[핵심 발견점]\n",
        " 1. 정밀 규격 로트의 평균 확정 불량률 6.6%, 일반 규격 2.8%\n",
        " 2. 규격 불일치 로트 8.1% vs 일치 3.9%\n",
        " 3. 납기 14일 이상 지연 로트 9.4% vs 정시 납품 3.4%\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 04. 다변량 분석 및 머신러닝 기반 은닉 품질 결함 패턴 식별 모델 고도화\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "다변량 분석 기반 은닉 품질 결함 패턴 식별 모델을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **다변량 분석으로 단일 요인에서 드러나지 않는 은닉 품질 결함 패턴(규격 x 납기 상호작용, 업체 품질 추세)을 식별하는 피처와 기준 모델을 만든다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **표본 불량률, 지연 비율, 정밀 규격 x 촉박 납기, 치수 편차/공차, 업체 최근 추세 등 파생 피처 10개 이상 작성하여 원인을 분석하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/features/qlt02_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": [
        "\"\"\"규격·구매요구·납기·검사 정보로 파생 피처를 만들고 다변량 선형 기준 모델(Ridge)로 로트 확정 불량률을 예측한다. 시계열 교차검증으로 과적합을 점검하고, 상호작용·추세 피처로 단일 요인 분석에서 보이지 않던 은닉 결함 패턴을 찾는다.\"\"\"\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.pipeline import make_pipeline\n",
        "from sklearn.preprocessing import StandardScaler\n",
        "from sklearn.model_selection import TimeSeriesSplit\n",
        "from sklearn.inspection import permutation_importance\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "\n",
        "def mape(y, p):\n",
        "    return float(np.mean(np.abs(y - p) / y))\n",
        "\n",
        "\n",
        "d, X = make_features(make_lots(1200, seed=42))\n",
        "y = np.log(d[\"confirmed_rate\"].values)                    # 불량률은 치우친 분포라 로그로 학습\n",
        "print(f\"파생 피처 {X.shape[1]}개 생성 (목표 10개 이상: {'달성' if X.shape[1] >= 10 else '미달'})\")\n",
        "print(\"피처:\", \", \".join(X.columns))\n",
        "\n",
        "# 1) 시계열 교차검증(과거로 학습 -> 이후 기간 검증)\n",
        "model = make_pipeline(StandardScaler(), Ridge(alpha=1.0))\n",
        "rows = []\n",
        "for k, (tr, te) in enumerate(TimeSeriesSplit(n_splits=4).split(X), 1):\n",
        "    model.fit(X.iloc[tr], y[tr])\n",
        "    rows.append({\"폴드\": k, \"학습 로트\": len(tr), \"검증 로트\": len(te),\n",
        "                 \"학습 MAPE\": round(mape(np.exp(y[tr]), np.exp(model.predict(X.iloc[tr]))), 3),\n",
        "                 \"검증 MAPE\": round(mape(np.exp(y[te]), np.exp(model.predict(X.iloc[te]))), 3)})\n",
        "cv = pd.DataFrame(rows)\n",
        "print(\"\\n[시계열 교차검증]\")\n",
        "print(cv.to_string(index=False))\n",
        "m = cv[\"검증 MAPE\"].mean()\n",
        "print(f\"기준 모델 평균 검증 MAPE {m:.1%} (목표 25% 이하: {'달성' if m <= 0.25 else '미달'}), 학습-검증 차이 {(cv['검증 MAPE'] - cv['학습 MAPE']).mean():+.1%}p\")\n",
        "\n",
        "# 2) 피처 중요도(순열 중요도)로 차원 축소 후보 확인\n",
        "tr = d[\"month\"] < 18\n",
        "model.fit(X[tr], y[tr])\n",
        "imp = permutation_importance(model, X[~tr], y[~tr], n_repeats=5, random_state=0)\n",
        "fi = pd.Series(imp.importances_mean, index=X.columns).sort_values(ascending=False)\n",
        "print(\"\\n[순열 중요도 상위 8개]\")\n",
        "print(fi.head(8).round(4).to_string())\n",
        "low = list(fi[fi < 0.002].index)\n",
        "Xr = X.drop(columns=low)\n",
        "model.fit(Xr[tr], y[tr])\n",
        "print(f\"중요도 0.002 미만 {len(low)}개 제거 후 검증 MAPE {mape(np.exp(y[~tr]), np.exp(model.predict(Xr[~tr]))):.1%} (피처 {Xr.shape[1]}개)\")\n",
        "\n",
        "# 3) 은닉 결함 패턴: 단일 요인으로는 약하지만 조합될 때 커지는 구간\n",
        "piv = d.assign(규격=np.where(X[\"precision\"] == 1, \"정밀\", \"일반\"), 납기=np.where(X[\"short_lead\"] == 1, \"촉박(<45일)\", \"여유\")).pivot_table(\n",
        "    index=\"규격\", columns=\"납기\", values=\"confirmed_rate\", aggfunc=\"mean\")\n",
        "print(\"\\n[은닉 패턴 1] 규격 x 납기 평균 확정 불량률\")\n",
        "print((piv * 100).round(1).astype(str).add(\"%\").to_string())\n",
        "lift = piv.loc[\"정밀\", \"촉박(<45일)\"] / piv.loc[\"일반\", \"여유\"]\n",
        "print(f\" -> 정밀 규격을 촉박한 납기로 계약하면 불량률이 {lift:.1f}배: 구매요구 단계에서 납기 여유를 확보해야 한다.\")\n",
        "drift = d[d[\"vendor\"] == \"V10\"].groupby(d[\"month\"] >= 18)[\"confirmed_rate\"].mean()\n",
        "print(f\"[은닉 패턴 2] V10 업체 불량률 전반기 {drift[False]:.1%} -> 후반기 {drift[True]:.1%} (최근 추세 피처 vendor_trend로 포착)\")\n",
        "no_int = make_pipeline(StandardScaler(), Ridge(1.0)).fit(X.drop(columns=[\"prec_x_short\", \"vendor_trend\"])[tr], y[tr])\n",
        "print(f\"상호작용·추세 피처 제외 시 검증 MAPE {mape(np.exp(y[~tr]), np.exp(no_int.predict(X.drop(columns=['prec_x_short', 'vendor_trend'])[~tr]))):.1%} -> 포함 시 {mape(np.exp(y[~tr]), np.exp(make_pipeline(StandardScaler(), Ridge(1.0)).fit(X[tr], y[tr]).predict(X[~tr]))):.1%}\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(12, 4))\n",
        "fi.head(10)[::-1].plot.barh(ax=ax[0], color=\"tab:blue\"); ax[0].set_title(\"Permutation importance (Ridge baseline)\")\n",
        "p = np.exp(make_pipeline(StandardScaler(), Ridge(1.0)).fit(X[tr], y[tr]).predict(X[~tr]))\n",
        "ax[1].scatter(np.exp(y[~tr]), p, s=8, alpha=0.6); ax[1].plot([0, 0.4], [0, 0.4], \"k--\")\n",
        "ax[1].set_xlabel(\"confirmed defect rate\"); ax[1].set_ylabel(\"predicted\"); ax[1].set_title(\"Baseline prediction (hold-out months 18-23)\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "파생 피처 20개 생성 (목표 10개 이상: 달성)\n",
        "피처: obs_rate, log_lot, delay_pos, delay_ratio, short_lead, precision, prec_x_short, spec_risk, dim_ratio, vendor_hist, vendor_trend, urgent, fill_ratio, dim_share, fsc_5305, fsc_5330, fsc_5935, fsc_6135, fsc_2530, fsc_8415\n",
        "\n",
        "[시계열 교차검증]\n",
        " 폴드  학습 로트  검증 로트  학습 MAPE  검증 MAPE\n",
        "  1    240    240    0.211    0.204\n",
        "  2    480    240    0.197    0.184\n",
        "  3    720    240    0.191    0.167\n",
        "  4    960    240    0.182    0.199\n",
        "기준 모델 평균 검증 MAPE 18.9% (목표 25% 이하: 달성), 학습-검증 차이 -0.7%p\n",
        "\n",
        "[순열 중요도 상위 8개]\n",
        "vendor_hist     0.3258\n",
        "delay_pos       0.2496\n",
        "spec_risk       0.1612\n",
        "precision       0.1299\n",
        "prec_x_short    0.0878\n",
        "obs_rate        0.0634\n",
        "urgent          0.0443\n",
        "fsc_5305        0.0283\n",
        "중요도 0.002 미만 7개 제거 후 검증 MAPE 20.1% (피처 13개)\n",
        "\n",
        "[은닉 패턴 1] 규격 x 납기 평균 확정 불량률\n",
        "납기    여유 촉박(<45일)\n",
        "규격               \n",
        "일반  2.7%     3.5%\n",
        "정밀  5.0%    15.0%\n",
        " -> 정밀 규격을 촉박한 납기로 계약하면 불량률이 5.7배: 구매요구 단계에서 납기 여유를 확보해야 한다.\n",
        "[은닉 패턴 2] V10 업체 불량률 전반기 6.3% -> 후반기 11.1% (최근 추세 피처 vendor_trend로 포착)\n",
        "상호작용·추세 피처 제외 시 검증 MAPE 23.0% -> 포함 시 20.0%\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 05. 로트별 품질 위험 지수(Quality Risk Score) 산출 및 조기경보 엔진 개발\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "로트별 품질 위험 지수 산출 및 조기경보 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **앙상블 모델로 로트별 확정 불량률을 예측해 품질 위험 지수(0~100)와 경보 등급을 산출하는 조기경보 엔진을 개발한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **그래디언트 부스팅과 랜덤 포레스트를 묶은 앙상블을 앞 18개월로 학습하고 최근 6개월 hold-out에서 기준 모델과 MAE·RMSE·R2를 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/models/qlt02_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": [
        "\"\"\"그래디언트 부스팅과 랜덤 포레스트 앙상블로 로트 확정 불량률을 예측해 기준 모델(Ridge) 대비 오차 개선과 R2를 검증하고, 예측 불량률을 '허용 불량률 초과 확률' 기반 품질 위험 지수(0~100)와 경보 등급으로 바꾸는 조기경보 엔진을 만든다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy import stats\n",
        "from sklearn.linear_model import Ridge\n",
        "from sklearn.pipeline import make_pipeline\n",
        "from sklearn.preprocessing import StandardScaler\n",
        "from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor\n",
        "from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "\n",
        "class RiskEngine:\n",
        "    \"\"\"앙상블(GBM 0.7 + RF 0.3, 로그 불량률) -> 위험 지수 = P(확정 불량률 > 5%) x 100\"\"\"\n",
        "\n",
        "    def __init__(self, seed=0):\n",
        "        self.gbm = GradientBoostingRegressor(n_estimators=600, max_depth=2, learning_rate=0.05, subsample=0.8, random_state=seed)\n",
        "        self.rf = RandomForestRegressor(n_estimators=300, min_samples_leaf=2, random_state=seed)\n",
        "\n",
        "    def fit(self, X, rate):\n",
        "        y = np.log(rate)\n",
        "        self.gbm.fit(X, y); self.rf.fit(X, y)\n",
        "        self.sigma = float(np.std(y - self._log_pred(X)) * 1.5)        # 학습 잔차는 작게 나오므로 1.5배로 보수적 설정\n",
        "        return self\n",
        "\n",
        "    def _log_pred(self, X):\n",
        "        return 0.7 * self.gbm.predict(X) + 0.3 * self.rf.predict(X)\n",
        "\n",
        "    def predict_rate(self, X):\n",
        "        return np.exp(self._log_pred(X))\n",
        "\n",
        "    def risk(self, X):\n",
        "        z = (np.log(BAD_RATE) - self._log_pred(X)) / self.sigma\n",
        "        return np.round(100 * (1 - stats.norm.cdf(z)), 1)\n",
        "\n",
        "\n",
        "def grade(score):\n",
        "    return np.where(score >= 70, \"경보\", np.where(score >= 40, \"주의\", \"정상\"))\n",
        "\n",
        "\n",
        "d, X = make_features(make_lots(1200, seed=42))\n",
        "tr, te = (d[\"month\"] < 18).values, (d[\"month\"] >= 18).values          # 앞 18개월 학습, 뒤 6개월 hold-out\n",
        "y = d[\"confirmed_rate\"].values\n",
        "\n",
        "base = make_pipeline(StandardScaler(), Ridge(1.0)).fit(X[tr], np.log(y[tr]))\n",
        "pb = np.exp(base.predict(X[te]))\n",
        "eng = RiskEngine().fit(X[tr], y[tr])\n",
        "pe = eng.predict_rate(X[te])\n",
        "\n",
        "res = pd.DataFrame([{\"모델\": name, \"MAE(%p)\": round(mean_absolute_error(y[te], p) * 100, 3), \"RMSE(%p)\": round(np.sqrt(mean_squared_error(y[te], p)) * 100, 3),\n",
        "                     \"R2\": round(r2_score(y[te], p), 3), \"MAPE\": round(float(np.mean(np.abs(y[te] - p) / y[te])), 3)}\n",
        "                    for name, p in [(\"업체 평균(단순)\", d[\"vendor_hist_rate\"].values[te]), (\"기준 모델(Ridge)\", pb), (\"앙상블 엔진\", pe)]])\n",
        "print(\"[Hold-out(최근 6개월) 성능 비교]\")\n",
        "print(res.to_string(index=False))\n",
        "mae_imp = 1 - res.loc[2, \"MAE(%p)\"] / res.loc[1, \"MAE(%p)\"]\n",
        "rmse_imp = 1 - res.loc[2, \"RMSE(%p)\"] / res.loc[1, \"RMSE(%p)\"]\n",
        "print(f\"기준 모델 대비 MAE {mae_imp:.1%}, RMSE {rmse_imp:.1%} 개선 (목표 20% 이상: {'달성' if max(mae_imp, rmse_imp) >= 0.2 else '미달'}), \"\n",
        "      f\"R2 {res.loc[2, 'R2']:.3f} (목표 0.85 이상: {'달성' if res.loc[2, 'R2'] >= 0.85 else '미달'})\")\n",
        "\n",
        "# 조기경보: 위험 지수와 등급\n",
        "h = d[te].copy()\n",
        "h[\"risk\"] = eng.risk(X[te]); h[\"grade\"] = grade(h[\"risk\"])\n",
        "h[\"base_accept\"] = h[\"defects\"] <= h[\"accept_c\"]\n",
        "print(\"\\n[위험 등급별 실제 불량 로트 비율]\")\n",
        "print(h.groupby(\"grade\").agg(로트수=(\"lot_id\", \"size\"), 실제불량=(\"bad_lot\", \"sum\"), 불량비율=(\"bad_lot\", \"mean\")).reindex([\"경보\", \"주의\", \"정상\"]).round(3).to_string())\n",
        "flag = h[\"grade\"] != \"정상\"\n",
        "rec = (flag & (h[\"bad_lot\"] == 1)).sum() / h[\"bad_lot\"].sum()\n",
        "missed_by_sampling = h[(h[\"base_accept\"]) & (h[\"bad_lot\"] == 1)]\n",
        "print(f\"\\n경보+주의 로트가 실제 불량 로트를 잡아낸 비율 {rec:.1%}\")\n",
        "print(f\"샘플링 판정에서 오합격된 불량 로트 {len(missed_by_sampling)}건 중 조기경보가 잡은 건 {(missed_by_sampling['grade'] != '정상').sum()}건 -> 정밀 재검사 대상으로 넘긴다\")\n",
        "\n",
        "# 안정성: 같은 입력 반복 산출 일관성, 앙상블 두 모델 간 편차\n",
        "again = RiskEngine().fit(X[tr], y[tr]).risk(X[te])\n",
        "print(f\"위험 지수 재산출 일관성 {np.mean(again == h['risk'].values):.0%}, GBM-RF 예측 불일치(로그 표준편차) {np.std(eng.gbm.predict(X[te]) - eng.rf.predict(X[te])):.3f}\")\n",
        "print(\"\\n[위험 지수 상위 5개 로트]\")\n",
        "print(h.sort_values(\"risk\", ascending=False).head(5)[[\"lot_id\", \"vendor\", \"fsc\", \"spec_class\", \"delay_days\", \"risk\", \"grade\", \"confirmed_rate\"]].to_string(index=False))\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(12, 4))\n",
        "ax[0].scatter(y[te], pb, s=8, alpha=0.5, label=\"Ridge baseline\"); ax[0].scatter(y[te], pe, s=8, alpha=0.6, label=\"ensemble\")\n",
        "ax[0].plot([0, 0.4], [0, 0.4], \"k--\"); ax[0].set_xlabel(\"confirmed rate\"); ax[0].set_ylabel(\"predicted\"); ax[0].legend(); ax[0].set_title(\"Hold-out prediction\")\n",
        "for g, c in [(0, \"tab:blue\"), (1, \"tab:red\")]:\n",
        "    ax[1].hist(h[h[\"bad_lot\"] == g][\"risk\"], bins=20, alpha=0.6, color=c, label=[\"good lot\", \"bad lot\"][g])\n",
        "ax[1].axvline(40, ls=\"--\", c=\"orange\"); ax[1].axvline(70, ls=\"--\", c=\"red\"); ax[1].legend(); ax[1].set_title(\"Quality risk score (0-100)\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[Hold-out(최근 6개월) 성능 비교]\n",
        "          모델  MAE(%p)  RMSE(%p)    R2  MAPE\n",
        "   업체 평균(단순)    2.734     4.835 0.182 0.881\n",
        "기준 모델(Ridge)    0.988     2.367 0.804 0.200\n",
        "      앙상블 엔진    0.756     1.296 0.941 0.188\n",
        "기준 모델 대비 MAE 23.5%, RMSE 45.2% 개선 (목표 20% 이상: 달성), R2 0.941 (목표 0.85 이상: 달성)\n",
        "\n",
        "[위험 등급별 실제 불량 로트 비율]\n",
        "       로트수  실제불량   불량비율\n",
        "grade                  \n",
        "경보      61    58  0.951\n",
        "주의      16     9  0.562\n",
        "정상     227     5  0.022\n",
        "\n",
        "경보+주의 로트가 실제 불량 로트를 잡아낸 비율 93.1%\n",
        "샘플링 판정에서 오합격된 불량 로트 15건 중 조기경보가 잡은 건 10건 -> 정밀 재검사 대상으로 넘긴다\n",
        "위험 지수 재산출 일관성 100%, GBM-RF 예측 불일치(로그 표준편차) 0.177\n",
        "\n",
        "[위험 지수 상위 5개 로트]\n",
        "     lot_id vendor  fsc spec_class  delay_days  risk grade  confirmed_rate\n",
        "LOT42-00958    V09 5935  precision          -4 100.0    경보          0.1037\n",
        "LOT42-00469    V10 5935  precision          -3 100.0    경보          0.3054\n",
        "LOT42-01133    V11 2530  precision           6 100.0    경보          0.3937\n",
        "LOT42-00827    V12 2530  precision          -1 100.0    경보          0.1025\n",
        "LOT42-00303    V11 2530  precision          -2 100.0    경보          0.1144\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 06. 수입 검사 불량 추세 및 고위험 납품업체 모니터링 품질 대시보드 구현\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "수입 검사 불량 추세 및 고위험 납품업체 모니터링 대시보드를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **수입 검사 불량 추세와 고위험 납품업체를 모니터링하는 품질 대시보드를 만들고, 경보 로트마다 기여도 상위 3개 인자를 설명한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **월별 표본 불량률과 관리한계, 업체별 평균 위험 지수와 추세(악화·유지), 위험 지수 분포, 대표 로트의 기여도 워터폴을 2x2 대시보드로 그린다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/explainability/qlt02_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": [
        "\"\"\"수령검사 불량 추세와 고위험 납품업체를 모니터링하는 품질 대시보드를 만들고, 경보 로트마다 7개 요인군의 정확한 섀플리 기여도를 계산해 기여도 상위 3개 인자와 자연어 설명·조치 제안을 자동 생성한다.\"\"\"\n",
        "import itertools\n",
        "import math\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy import stats\n",
        "from sklearn.ensemble import GradientBoostingRegressor\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "GROUPS = {\"업체 이력\": [\"vendor_hist\", \"vendor_trend\"], \"규격 변경·불일치\": [\"spec_risk\"], \"정밀 규격\": [\"precision\"],\n",
        "          \"납기 지연·촉박\": [\"delay_pos\", \"delay_ratio\", \"short_lead\", \"prec_x_short\"],\n",
        "          \"표본검사 결과\": [\"obs_rate\", \"dim_ratio\", \"dim_share\"], \"구매요구 조건\": [\"urgent\", \"fill_ratio\", \"log_lot\"],\n",
        "          \"품목군\": [f\"fsc_{f}\" for f in FSC]}\n",
        "ACTION = {\"업체 이력\": \"업체 품질 이력 재점검·시정조치 요구\", \"규격 변경·불일치\": \"구매요구 규격서와 납품 규격 대조\",\n",
        "          \"정밀 규격\": \"정밀 측정 장비로 치수 재측정\", \"납기 지연·촉박\": \"납기 여유 확보·지연 사유 확인\",\n",
        "          \"표본검사 결과\": \"표본 크기 확대 재검사\", \"구매요구 조건\": \"긴급 구매 건 사후 검사 강화\", \"품목군\": \"품목군별 검사 기준 확인\"}\n",
        "\n",
        "d, X = make_features(make_lots(1200, seed=42))\n",
        "tr, te = (d[\"month\"] < 18).values, (d[\"month\"] >= 18).values\n",
        "y = np.log(d[\"confirmed_rate\"].values)\n",
        "gbm = GradientBoostingRegressor(n_estimators=300, max_depth=2, learning_rate=0.08, subsample=0.8, random_state=0).fit(X.values[tr], y[tr])\n",
        "sigma = float(np.std(y[tr] - gbm.predict(X.values[tr])) * 1.5)\n",
        "h = d[te].copy()\n",
        "h[\"pred\"] = np.exp(gbm.predict(X.values[te]))\n",
        "h[\"risk\"] = np.round(100 * (1 - stats.norm.cdf((np.log(BAD_RATE) - np.log(h[\"pred\"])) / sigma)), 1)\n",
        "h[\"grade\"] = np.where(h[\"risk\"] >= 70, \"경보\", np.where(h[\"risk\"] >= 40, \"주의\", \"정상\"))\n",
        "\n",
        "# ---------------- 1) 대시보드 요약 표 ----------------\n",
        "print(\"[품질 대시보드] 최근 6개월 수령검사 현황 [교육용가상자료]\")\n",
        "m = d.groupby(\"month\").agg(sample_rate=(\"defects\", \"sum\"), n=(\"sample_n\", \"sum\"), conf=(\"confirmed_rate\", \"mean\"))\n",
        "m[\"sample_rate\"] = m[\"sample_rate\"] / m[\"n\"]\n",
        "p0 = m.loc[:17, \"sample_rate\"].mean()\n",
        "ucl = p0 + 3 * np.sqrt(p0 * (1 - p0) / m[\"n\"])\n",
        "print(f\"월별 표본 불량률 관리한계 초과 월: {list(m.index[m['sample_rate'] > ucl])} (기준 {p0:.2%})\")\n",
        "vend = h.groupby(\"vendor\").agg(로트=(\"lot_id\", \"size\"), 평균위험=(\"risk\", \"mean\"), 경보=(\"grade\", lambda s: (s == \"경보\").sum()),\n",
        "                               실제불량률=(\"bad_lot\", \"mean\"), 최근추세=(\"vendor_recent_rate\", \"mean\"), 과거=(\"vendor_hist_rate\", \"mean\"))\n",
        "vend[\"추세\"] = np.where(vend[\"최근추세\"] > vend[\"과거\"] * 1.2, \"악화\", \"유지\")\n",
        "vend[\"상태\"] = np.where(vend[\"평균위험\"] >= 50, \"고위험\", np.where(vend[\"평균위험\"] >= 25, \"관찰\", \"정상\"))\n",
        "print(vend.sort_values(\"평균위험\", ascending=False).head(6)[[\"로트\", \"평균위험\", \"경보\", \"실제불량률\", \"추세\", \"상태\"]].round(2).to_string())\n",
        "\n",
        "# ---------------- 2) 정확한 섀플리 기여도(요인군 7개, 2^7 연합) ----------------\n",
        "cols = {g: [X.columns.get_loc(c) for c in cs] for g, cs in GROUPS.items()}\n",
        "names = list(GROUPS)\n",
        "bg = X[tr].sample(40, random_state=1).values.astype(float)                       # 배경 데이터(학습 구간 표본)\n",
        "base_val = gbm.predict(bg).mean()\n",
        "coal = list(itertools.product([0, 1], repeat=len(names)))\n",
        "\n",
        "\n",
        "def shapley(x):\n",
        "    \"\"\"연합 S마다 S에 속한 요인군은 x 값, 나머지는 배경 값으로 바꿔 평균 예측 v(S)를 구하고 섀플리 공식으로 합산.\"\"\"\n",
        "    batch = np.repeat(bg[None], len(coal), axis=0)\n",
        "    for k, s in enumerate(coal):\n",
        "        for g, on in zip(names, s):\n",
        "            if on:\n",
        "                batch[k][:, cols[g]] = x[cols[g]]\n",
        "    v = dict(zip(coal, gbm.predict(batch.reshape(-1, bg.shape[1])).reshape(len(coal), -1).mean(1)))\n",
        "    n, phi = len(names), np.zeros(len(names))\n",
        "    for s in coal:\n",
        "        for i in range(n):\n",
        "            if not s[i]:\n",
        "                w = math.factorial(sum(s)) * math.factorial(n - sum(s) - 1) / math.factorial(n)\n",
        "                phi[i] += w * (v[s[:i] + (1,) + s[i + 1:]] - v[s])\n",
        "    return phi, v[tuple([1] * n)]\n",
        "\n",
        "\n",
        "alert = h[h[\"grade\"] == \"경보\"]\n",
        "rows = []\n",
        "for idx in alert.index:\n",
        "    phi, full = shapley(X.loc[idx].values)\n",
        "    top = np.argsort(-phi)[:3]\n",
        "    rows.append({\"lot_id\": h.loc[idx, \"lot_id\"], \"risk\": h.loc[idx, \"risk\"], **{names[i]: phi[i] for i in range(len(names))},\n",
        "                 \"top3\": [names[i] for i in top], \"top3_phi\": [phi[i] for i in top], \"add_err\": abs(base_val + phi.sum() - full)})\n",
        "ex = pd.DataFrame(rows)\n",
        "done = (ex[\"top3\"].map(len) == 3).mean()\n",
        "print(f\"\\n경보 로트 {len(ex)}건 설명 생성: 상위 3개 인자 자동 산출률 {done:.0%} (목표 100%: {'달성' if done == 1 else '미달'}), \"\n",
        "      f\"가법성 오차 최대 {ex['add_err'].max():.1e}\")\n",
        "print(\"[요인군별 평균 기여도(로그 불량률, 경보 로트)]\")\n",
        "print(ex[names].mean().sort_values(ascending=False).round(3).to_string())\n",
        "\n",
        "print(\"\\n[자동 설명 예시]\")\n",
        "for r in ex.sort_values(\"risk\", ascending=False).head(3).itertuples():\n",
        "    parts = \", \".join(f\"{g}(x{np.exp(p):.2f})\" for g, p in zip(r.top3, r.top3_phi))\n",
        "    print(f\"  {r.lot_id} 위험 {r.risk:.0f}: {parts} 순으로 불량률 예측을 높였다 -> 조치: {ACTION[r.top3[0]]}\")\n",
        "\n",
        "# 설명 타당성 점검: 규격 불일치 로트에서 '규격' 요인이 양(+)으로 나오는가\n",
        "mm = alert[\"spec_mismatch\"].values == 1\n",
        "if mm.any():\n",
        "    print(f\"타당성 점검: 규격 불일치 경보 로트 {mm.sum()}건 중 '규격 변경·불일치' 기여 양수 {np.mean(ex.loc[mm, '규격 변경·불일치'] > 0):.0%}\")\n",
        "\n",
        "fig, ax = plt.subplots(2, 2, figsize=(13, 8))\n",
        "ax[0, 0].plot(m.index, m[\"sample_rate\"], \"o-\", label=\"sample defect rate\"); ax[0, 0].plot(m.index, ucl, \"r--\", label=\"UCL (3 sigma)\")\n",
        "ax[0, 0].axvspan(17.5, 23.5, color=\"gray\", alpha=0.1); ax[0, 0].legend(); ax[0, 0].set_title(\"Monthly inspection defect trend\"); ax[0, 0].set_xlabel(\"month\")\n",
        "vs = vend.sort_values(\"평균위험\")\n",
        "ax[0, 1].barh(vs.index, vs[\"평균위험\"], color=[\"tab:red\" if s == \"고위험\" else \"tab:orange\" if s == \"관찰\" else \"tab:green\" for s in vs[\"상태\"]])\n",
        "ax[0, 1].set_title(\"Vendor mean risk score (last 6 months)\")\n",
        "ax[1, 0].hist(h[\"risk\"], bins=25, color=\"tab:blue\"); ax[1, 0].axvline(40, c=\"orange\", ls=\"--\"); ax[1, 0].axvline(70, c=\"red\", ls=\"--\")\n",
        "ax[1, 0].set_title(\"Lot risk score distribution\")\n",
        "r0 = ex.sort_values(\"risk\", ascending=False).iloc[0]\n",
        "vals = [r0[g] for g in names]; order = np.argsort(vals)[::-1]\n",
        "cum = base_val\n",
        "eng = [\"vendor history\", \"spec change\", \"precision spec\", \"delivery delay\", \"sample result\", \"PR condition\", \"item group\"]\n",
        "for k, i in enumerate(order):\n",
        "    ax[1, 1].barh(k, vals[i], left=cum, color=\"tab:red\" if vals[i] > 0 else \"tab:blue\"); cum += vals[i]\n",
        "ax[1, 1].set_yticks(range(len(order))); ax[1, 1].set_yticklabels([eng[i] for i in order]); ax[1, 1].invert_yaxis()\n",
        "ax[1, 1].set_xlabel(\"log defect rate\"); ax[1, 1].set_title(f\"Waterfall: {r0['lot_id']}\")\n",
        "plt.suptitle(\"QLT02 incoming inspection quality dashboard (synthetic)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[품질 대시보드] 최근 6개월 수령검사 현황 [교육용가상자료]\n",
        "월별 표본 불량률 관리한계 초과 월: [1, 11, 13, 18] (기준 4.09%)\n",
        "        로트   평균위험  경보  실제불량률  추세   상태\n",
        "vendor                               \n",
        "V12     19  72.55  12   0.79  유지  고위험\n",
        "V10     16  55.41   8   0.69  악화  고위험\n",
        "V11     22  49.74   8   0.59  악화   관찰\n",
        "V07     14  28.34   3   0.21  유지   관찰\n",
        "V09     21  26.50   5   0.29  유지   관찰\n",
        "V06     31  21.46   5   0.19  유지   정상\n",
        "\n",
        "경보 로트 59건 설명 생성: 상위 3개 인자 자동 산출률 100% (목표 100%: 달성), 가법성 오차 최대 6.7e-16\n",
        "[요인군별 평균 기여도(로그 불량률, 경보 로트)]\n",
        "표본검사 결과      0.615\n",
        "납기 지연·촉박     0.216\n",
        "정밀 규격        0.165\n",
        "업체 이력        0.116\n",
        "품목군          0.053\n",
        "구매요구 조건      0.049\n",
        "규격 변경·불일치    0.038\n",
        "\n",
        "[자동 설명 예시]\n",
        "  LOT42-00172 위험 100: 표본검사 결과(x3.45), 납기 지연·촉박(x2.22), 정밀 규격(x1.33) 순으로 불량률 예측을 높였다 -> 조치: 표본 크기 확대 재검사\n",
        "  LOT42-01133 위험 100: 표본검사 결과(x3.37), 납기 지연·촉박(x1.84), 정밀 규격(x1.28) 순으로 불량률 예측을 높였다 -> 조치: 표본 크기 확대 재검사\n",
        "  LOT42-01072 위험 100: 표본검사 결과(x2.72), 업체 이력(x1.62), 정밀 규격(x1.26) 순으로 불량률 예측을 높였다 -> 조치: 표본 크기 확대 재검사\n",
        "타당성 점검: 규격 불일치 경보 로트 10건 중 '규격 변경·불일치' 기여 양수 100%\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 07. 합부 판정 경계선 품목에 대한 품질 검사관 정밀 재검사(HITL) 워크플로우 설계\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "합부 판정 경계선 품목에 대한 검사관 정밀 재검사(HITL) 워크플로우를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **합부 판정 경계선에 있는 로트를 자동 확정하지 않고 품질 검사관 정밀 재검사로 넘기는 인간 개입(HITL) 워크플로우를 설계하고, 한정된 검사 시간을 최적 배분한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **표본 불량 수가 합격판정수 근처이거나 위험 확률이 중간인 로트를 재검사 대기열에 등록 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/optimization/qlt02_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": [
        "\"\"\"샘플링 판정과 위험 지수가 경계선에 있는 로트를 자동 확정하지 않고 검사관 정밀 재검사 대기열로 보내는 인간 개입(HITL) 워크플로우를 설계하고, 한정된 검사관 시간을 정수계획(MIP)으로 배분해 현행 선착순 방식 대비 품질 비용 절감과 수요 급증 민감도를 시뮬레이션한다.\"\"\"\n",
        "import hashlib\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy import stats\n",
        "from scipy.optimize import milp, LinearConstraint, Bounds\n",
        "from sklearn.ensemble import GradientBoostingRegressor\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "# 교육용 가정 단가(원): 실제 계약·예산 자료 아님\n",
        "HOUR_COST = 50_000            # 검사관 1시간 비용\n",
        "MISS_FACTOR = 3.0             # 불량품이 부대로 불출될 때 단가 대비 처리비(교체·재수송·정비 손실) 배수\n",
        "REJECT_COST_RATE = 0.03       # 양품 로트를 잘못 반품할 때 로트 금액 대비 지연·재납품 비용\n",
        "CAP_HOURS = 24                # 월 정밀 재검사 가용 시간(검사관 2명 x 12시간)\n",
        "\n",
        "d, X = make_features(make_lots(1200, seed=42))\n",
        "tr, te = (d[\"month\"] < 18).values, (d[\"month\"] >= 18).values\n",
        "y = np.log(d[\"confirmed_rate\"].values)\n",
        "gbm = GradientBoostingRegressor(n_estimators=300, max_depth=2, learning_rate=0.08, subsample=0.8, random_state=0).fit(X.values[tr], y[tr])\n",
        "sigma = float(np.std(y[tr] - gbm.predict(X.values[tr])) * 1.5)\n",
        "h = d[te].copy()\n",
        "h[\"p_bad\"] = 1 - stats.norm.cdf((np.log(BAD_RATE) - gbm.predict(X.values[te])) / sigma)\n",
        "h[\"accept\"] = h[\"defects\"] <= h[\"accept_c\"]\n",
        "\n",
        "\n",
        "def costs(df):\n",
        "    value = df[\"lot_size\"] * df[\"unit_cost\"]\n",
        "    df[\"c_fa\"] = df[\"lot_size\"] * df[\"confirmed_rate\"] * df[\"unit_cost\"] * MISS_FACTOR        # 오합격 시 실제 비용\n",
        "    df[\"c_fr\"] = value * REJECT_COST_RATE                                                     # 오불합격 시 실제 비용\n",
        "    df[\"hours\"] = np.where(df[\"spec_class\"] == \"precision\", 4.0, 2.0) + df[\"lot_size\"] / 2000\n",
        "    # 재검사 전 기대 비용(모델 추정): 합격 판정이면 P(불량)x오합격비, 불합격 판정이면 (1-P)x오불합격비\n",
        "    exp_fa = df[\"p_bad\"] * df[\"lot_size\"] * np.exp(np.log(BAD_RATE) + 0.7) * df[\"unit_cost\"] * MISS_FACTOR\n",
        "    df[\"benefit\"] = np.where(df[\"accept\"], exp_fa, (1 - df[\"p_bad\"]) * df[\"c_fr\"]) - df[\"hours\"] * HOUR_COST\n",
        "    df[\"realized_no\"] = np.where(df[\"accept\"], df[\"bad_lot\"] * df[\"c_fa\"], (1 - df[\"bad_lot\"]) * df[\"c_fr\"])\n",
        "    return df\n",
        "\n",
        "\n",
        "# ---------------- 1) 경계선 로트 선별 -> 재검사 대기열(자동 확정 금지) ----------------\n",
        "h[\"margin\"] = h[\"defects\"] - h[\"accept_c\"]\n",
        "h[\"border\"] = (h[\"margin\"].abs() <= 1) | h[\"p_bad\"].between(0.3, 0.8) | (h[\"accept\"] & (h[\"p_bad\"] > 0.5))\n",
        "q = costs(h[h[\"border\"]].copy())\n",
        "print(f\"[1] 최근 6개월 로트 {len(h)}건 중 경계선 {len(q)}건 -> 검사관 재검사 대기열 등록, 자동 확정 0건 [교육용가상자료]\")\n",
        "print(f\"    선별 기준: |표본불량-합격판정수|<=1, 위험확률 30~80%, 합격 판정인데 위험확률>50%\")\n",
        "\n",
        "\n",
        "def select(df, policy, cap=CAP_HOURS):\n",
        "    \"\"\"월별 가용 시간 안에서 재검사 대상 선택. fifo=입고순, mip=기대 순편익 최대화 0-1 정수계획.\"\"\"\n",
        "    pick = np.zeros(len(df), bool)\n",
        "    for mth, g in df.groupby(\"month\"):\n",
        "        idx = np.where(df[\"month\"].values == mth)[0]\n",
        "        if policy == \"fifo\":\n",
        "            used = 0\n",
        "            for i in idx[np.argsort(df[\"receipt_date\"].values[idx])]:\n",
        "                if used + df[\"hours\"].values[i] <= cap:\n",
        "                    pick[i] = True; used += df[\"hours\"].values[i]\n",
        "        else:\n",
        "            b, w = df[\"benefit\"].values[idx], df[\"hours\"].values[idx]\n",
        "            res = milp(c=-b, constraints=[LinearConstraint(w[None, :], 0, cap)], integrality=np.ones(len(idx)), bounds=Bounds(0, 1))\n",
        "            pick[idx] = res.x > 0.5\n",
        "    return pick\n",
        "\n",
        "\n",
        "def total_cost(df, pick):\n",
        "    \"\"\"실제(확정) 결과로 계산: 재검사 로트는 검사비만(정밀 재검사로 정답 판정), 나머지는 샘플링 판정 오류 비용.\"\"\"\n",
        "    return float(np.sum(np.where(pick, df[\"hours\"] * HOUR_COST, df[\"realized_no\"])))\n",
        "\n",
        "\n",
        "res = {}\n",
        "for pol in (\"none\", \"fifo\", \"mip\"):\n",
        "    pick = np.zeros(len(q), bool) if pol == \"none\" else select(q, pol)\n",
        "    res[pol] = {\"재검사 로트\": int(pick.sum()), \"사용 시간\": round(float(q[\"hours\"][pick].sum()), 1), \"총 품질비용(백만원)\": round(total_cost(q, pick) / 1e6, 1),\n",
        "                \"놓친 불량 로트\": int(((~pick) & q[\"accept\"].values & (q[\"bad_lot\"].values == 1)).sum())}\n",
        "    if pol == \"mip\":\n",
        "        q[\"pick\"] = pick\n",
        "tab = pd.DataFrame(res).T.astype({\"재검사 로트\": int, \"놓친 불량 로트\": int}).rename(index={\"none\": \"재검사 없음\", \"fifo\": \"현행(선착순)\", \"mip\": \"최적화(MIP)\"})\n",
        "print(\"\\n[2] 재검사 배분 방식 비교 (월 가용 24시간)\")\n",
        "print(tab.to_string())\n",
        "cut = 1 - res[\"mip\"][\"총 품질비용(백만원)\"] / res[\"fifo\"][\"총 품질비용(백만원)\"]\n",
        "print(f\"현행 대비 품질비용 {cut:.1%} 절감 (목표 15% 이상: {'달성' if cut >= 0.15 else '미달'})\")\n",
        "\n",
        "# ---------------- 3) 검사관 결정 기록(HITL) ----------------\n",
        "log, prev, n_block = [], \"GENESIS\", 0\n",
        "INSP = [\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"]\n",
        "for i, r in enumerate(q[q[\"pick\"]].head(12).itertuples()):\n",
        "    who = INSP[(INSP.index(r.inspector) + 1 + i) % 4]\n",
        "    if who == r.inspector:                              # 최초 검사자는 재검사 불가(이해충돌)\n",
        "        n_block += 1; who = INSP[(INSP.index(who) + 1) % 4]\n",
        "    verdict = \"불합격\" if r.bad_lot else \"합격\"\n",
        "    reason = f\"전수 측정 확정불량률 {r.confirmed_rate:.1%}, 치수편차 {r.dim_dev_mm}mm\"\n",
        "    rec = f\"{r.lot_id}|{who}|{verdict}|{reason}|{prev}\"\n",
        "    prev = hashlib.sha256(rec.encode()).hexdigest()[:12]\n",
        "    log.append({\"로트\": r.lot_id, \"샘플링판정\": \"합격\" if r.accept else \"불합격\", \"위험확률\": round(r.p_bad, 2), \"재검사관\": who, \"최종\": verdict, \"근거\": reason, \"hash\": prev})\n",
        "L = pd.DataFrame(log)\n",
        "flip = (L[\"샘플링판정\"] != L[\"최종\"]).sum()\n",
        "print(f\"\\n[3] 검사관 재검사 기록 예시(12건 중 판정 번복 {flip}건, 근거 미기재 결정 0건, 최초 검사자 배정 차단 {n_block}건)\")\n",
        "print(L.head(5)[[\"로트\", \"샘플링판정\", \"위험확률\", \"재검사관\", \"최종\", \"근거\"]].to_string(index=False))\n",
        "\n",
        "# ---------------- 4) 민감도: 전시 전환 등 돌발 수요 ----------------\n",
        "print(\"\\n[4] 민감도 분석 (경계선 로트 증가, 검사 가용 시간 변화)\")\n",
        "rows = []\n",
        "for surge, capf in [(1.0, 1.0), (1.5, 1.0), (2.0, 1.0), (2.0, 0.7)]:\n",
        "    qq = q.sample(frac=surge, replace=surge > 1, random_state=7).reset_index(drop=True)\n",
        "    cf, cm = [total_cost(qq, select(qq, p, CAP_HOURS * capf)) for p in (\"fifo\", \"mip\")]\n",
        "    rows.append({\"물량 배수\": surge, \"가용시간\": f\"{capf:.0%}\", \"선착순(백만원)\": round(cf / 1e6, 1), \"MIP(백만원)\": round(cm / 1e6, 1), \"절감액(백만원)\": round((cf - cm) / 1e6, 1), \"절감률\": f\"{1 - cm / cf:.1%}\"})\n",
        "sens = pd.DataFrame(rows)\n",
        "print(sens.to_string(index=False))\n",
        "print(\"물량이 늘면 절감률은 낮아지지만 절감액은 커진다. 가용 시간이 부족할수록 기대 손실이 큰 로트부터 재검사해야 한다.\")\n",
        "print(\"최종 합부 판정은 검사관이 내리고, 모델은 재검사 순서만 제안한다.\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 3, figsize=(15, 4))\n",
        "ax[0].bar([\"none\", \"FIFO\", \"MIP\"], tab[\"총 품질비용(백만원)\"], color=[\"gray\", \"tab:orange\", \"tab:green\"]); ax[0].set_ylabel(\"quality cost (M KRW)\"); ax[0].set_title(\"Re-inspection allocation\")\n",
        "ax[1].scatter(q[\"hours\"], q[\"benefit\"] / 1e6, c=q[\"pick\"].map({True: \"tab:green\", False: \"lightgray\"}), s=14)\n",
        "ax[1].set_xlabel(\"re-inspection hours\"); ax[1].set_ylabel(\"expected net benefit (M KRW)\"); ax[1].set_title(\"MIP selection (green = selected)\")\n",
        "ax[2].plot(range(len(sens)), [float(s[:-1]) for s in sens[\"절감률\"]], \"o-\"); ax[2].set_xticks(range(len(sens)))\n",
        "ax[2].set_xticklabels([f\"x{a} / {b}\" for a, b in zip(sens[\"물량 배수\"], sens[\"가용시간\"])]); ax[2].set_ylabel(\"saving vs FIFO (%)\"); ax[2].set_title(\"Sensitivity: surge / capacity\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[1] 최근 6개월 로트 304건 중 경계선 57건 -> 검사관 재검사 대기열 등록, 자동 확정 0건 [교육용가상자료]\n",
        "    선별 기준: |표본불량-합격판정수|<=1, 위험확률 30~80%, 합격 판정인데 위험확률>50%\n",
        "\n",
        "[2] 재검사 배분 방식 비교 (월 가용 24시간)\n",
        "          재검사 로트  사용 시간  총 품질비용(백만원)  놓친 불량 로트\n",
        "재검사 없음         0    0.0         57.5        13\n",
        "현행(선착순)       35  137.7         25.7         5\n",
        "최적화(MIP)      32  121.3          8.8         3\n",
        "현행 대비 품질비용 65.8% 절감 (목표 15% 이상: 달성)\n",
        "\n",
        "[3] 검사관 재검사 기록 예시(12건 중 판정 번복 6건, 근거 미기재 결정 0건, 최초 검사자 배정 차단 3건)\n",
        "         로트 샘플링판정  위험확률  재검사관  최종                              근거\n",
        "LOT42-00127    합격  0.89 INS-2 불합격 전수 측정 확정불량률 5.7%, 치수편차 0.0523mm\n",
        "LOT42-00597    합격  0.11 INS-2  합격 전수 측정 확정불량률 3.3%, 치수편차 0.1563mm\n",
        "LOT42-00159    합격  0.18 INS-1  합격 전수 측정 확정불량률 2.8%, 치수편차 0.1883mm\n",
        "LOT42-01097    합격  0.08 INS-2  합격 전수 측정 확정불량률 2.6%, 치수편차 0.0287mm\n",
        "LOT42-00246    합격  0.33 INS-2 불합격   전수 측정 확정불량률 6.7%, 치수편차 0.21mm\n",
        "\n",
        "[4] 민감도 분석 (경계선 로트 증가, 검사 가용 시간 변화)\n",
        " 물량 배수 가용시간  선착순(백만원)  MIP(백만원)  절감액(백만원)   절감률\n",
        "   1.0 100%      25.7       8.8      16.9 65.9%\n",
        "   1.5 100%      92.1      51.7      40.5 43.9%\n",
        "   2.0 100%     123.6      77.9      45.7 36.9%\n",
        "   2.0  70%     145.9     100.5      45.4 31.1%\n",
        "물량이 늘면 절감률은 낮아지지만 절감액은 커진다. 가용 시간이 부족할수록 기대 손실이 큰 로트부터 재검사해야 한다.\n",
        "최종 합부 판정은 검사관이 내리고, 모델은 재검사 순서만 제안한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 08. 불량 부품 대량 납품 가상 주입 시 조기경보 신속성 스트레스 테스트\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "불량 부품 대량 납품 가상 주입 시 조기경보 스트레스 테스트를 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **평소 품질이 좋던 업체가 불량 부품을 대량 납품하는 상황을 가상으로 주입해 조기경보의 정확도와 신속성, 대시보드 갱신 성능을 시험한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **우수 업체 V03에 불량 로트 위주의 대량 납품 60로트와 손상 레코드 12건 주입하여 처리 지연 및 복원력을 시험하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/dashboard/qlt02_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": [
        "\"\"\"평소 품질이 좋던 가상 업체가 불량 로트를 대량 납품하는 상황과 손상 레코드를 주입해, 로트 경보 정확도·업체 단위 누적합(CUSUM) 경보의 신속성·대시보드 갱신 시간(물량 1~50배)을 측정하는 조기경보 스트레스 테스트를 수행한다.\"\"\"\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy import stats\n",
        "from sklearn.ensemble import GradientBoostingRegressor\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "base = make_lots(1200, seed=42)\n",
        "# ---------------- 1) 대량 불량 납품 주입 ----------------\n",
        "rng = np.random.default_rng(88)\n",
        "inj = make_lots(60, seed=88)\n",
        "inj[\"vendor\"] = \"V03\"                                            # 평소 우수 업체(숨은 품질 -0.3)\n",
        "inj[\"receipt_date\"] = pd.Timestamp(\"2025-09-01\") + pd.to_timedelta(np.sort(rng.integers(0, 25, 60)), unit=\"D\")\n",
        "inj[\"month\"] = 20\n",
        "inj[\"lot_size\"] = (inj[\"lot_size\"] * 2).astype(int)              # 대량 납품\n",
        "p = np.where(rng.random(60) < 0.8, rng.uniform(0.08, 0.25, 60), rng.uniform(0.01, 0.03, 60))   # 80%는 불량 로트\n",
        "inj[\"sample_n\"], inj[\"accept_c\"] = zip(*[sample_plan(s) for s in inj[\"lot_size\"]])\n",
        "inj[\"defects\"] = rng.binomial(inj[\"sample_n\"], p)\n",
        "inj[\"def_dim\"] = rng.binomial(inj[\"defects\"], 0.5); inj[\"def_spec\"] = inj[\"defects\"] - inj[\"def_dim\"]; inj[\"def_visual\"] = 0\n",
        "inj[\"dim_dev_mm\"] = (inj[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, 60)).round(4)\n",
        "inj[\"confirmed_rate\"] = p.round(4); inj[\"bad_lot\"] = (p > BAD_RATE).astype(int)\n",
        "inj[\"lot_id\"] = [f\"INJ-{i:03d}\" for i in range(60)]\n",
        "inj[\"injected\"] = 1\n",
        "base[\"injected\"] = 0\n",
        "allx = pd.concat([base, inj], ignore_index=True)\n",
        "\n",
        "# 손상 레코드 주입(측정치 결측, 음수 불량 수, 표본 0) -> 격리\n",
        "bad_idx = rng.choice(len(allx), 12, replace=False)\n",
        "allx.loc[bad_idx[:4], \"dim_dev_mm\"] = np.nan\n",
        "allx.loc[bad_idx[4:8], \"defects\"] = -1\n",
        "allx.loc[bad_idx[8:], \"sample_n\"] = 0\n",
        "\n",
        "\n",
        "def quarantine(df):\n",
        "    ok = df[\"dim_dev_mm\"].notna() & (df[\"defects\"] >= 0) & (df[\"sample_n\"] > 0) & (df[\"defects\"] <= df[\"sample_n\"])\n",
        "    return df[ok].copy(), df[~ok].copy()\n",
        "\n",
        "\n",
        "clean, quar = quarantine(allx)\n",
        "print(f\"[1] 주입: V03 업체 대량 납품 60로트(불량 {inj['bad_lot'].sum()}로트) + 손상 레코드 {len(bad_idx)}건 [교육용가상자료]\")\n",
        "print(f\"    손상 레코드 {len(quar)}건 격리(재입력 요청), 파이프라인 중단 0건\")\n",
        "\n",
        "d, X = make_features(clean)\n",
        "tr = (d[\"month\"] < 18).values\n",
        "y = np.log(d[\"confirmed_rate\"].values)\n",
        "gbm = GradientBoostingRegressor(n_estimators=300, max_depth=2, learning_rate=0.08, subsample=0.8, random_state=0).fit(X.values[tr], y[tr])\n",
        "sigma = float(np.std(y[tr] - gbm.predict(X.values[tr])) * 1.5)\n",
        "\n",
        "\n",
        "def score(Xv):\n",
        "    return 100 * (1 - stats.norm.cdf((np.log(BAD_RATE) - gbm.predict(Xv)) / sigma))\n",
        "\n",
        "\n",
        "# ---------------- 2) 로트 경보 정확도 ----------------\n",
        "win = (d[\"month\"] >= 18).values\n",
        "w = d[win].copy(); w[\"risk\"] = score(X.values[win])\n",
        "w[\"alert\"] = (w[\"risk\"] >= 70) | (w[\"defects\"] > w[\"accept_c\"])          # 위험 지수 경보 또는 샘플링 불합격\n",
        "acc = (w[\"alert\"] == (w[\"bad_lot\"] == 1)).mean()\n",
        "wi = w[w[\"injected\"] == 1]\n",
        "tp, fp = int((w[\"alert\"] & (w[\"bad_lot\"] == 1)).sum()), int((w[\"alert\"] & (w[\"bad_lot\"] == 0)).sum())\n",
        "print(f\"\\n[2] 스트레스 구간 로트 {len(w)}건: 경보 발령 정확도 {acc:.1%} (목표 95% 이상: {'달성' if acc >= 0.95 else '미달'})\")\n",
        "print(f\"    주입 불량 로트 탐지 {int((wi['alert'] & (wi['bad_lot'] == 1)).sum())}/{int(wi['bad_lot'].sum())}, 전체 정탐 {tp}건, 오경보 {fp}건\")\n",
        "acc_i = (wi[\"alert\"] == (wi[\"bad_lot\"] == 1)).mean()\n",
        "print(f\"    대량 납품 주입 로트 {len(wi)}건 경보 정확도 {acc_i:.1%} (목표 95% 이상: {'달성' if acc_i >= 0.95 else '미달'})\")\n",
        "print(f\"    샘플링 판정만 사용 시 정확도 {((w['defects'] > w['accept_c']) == (w['bad_lot'] == 1)).mean():.1%}\")\n",
        "\n",
        "# ---------------- 3) 업체 단위 CUSUM 경보 신속성 ----------------\n",
        "v3 = d[d[\"vendor\"] == \"V03\"].sort_values(\"receipt_date\")\n",
        "ref = v3[v3[\"month\"] < 18]\n",
        "p0 = (ref[\"defects\"].sum() / ref[\"sample_n\"].sum())\n",
        "first = v3[v3[\"injected\"] == 1][\"receipt_date\"].min()\n",
        "k, hlim, s, alarm, false_alarm = p0 * 2, 0.15, 0.0, None, 0\n",
        "trace = []\n",
        "for r in v3.itertuples():\n",
        "    s = max(0.0, s + r.defects / r.sample_n - k)       # 기준의 2배를 넘는 표본 불량률만 누적\n",
        "    trace.append(s)\n",
        "    if s > hlim and r.receipt_date < first:\n",
        "        false_alarm += 1; s = 0.0                        # 평시 오경보: 기록 후 초기화\n",
        "    if alarm is None and s > hlim and r.injected == 1:\n",
        "        alarm = r\n",
        "n_before = int((v3[v3[\"injected\"] == 1][\"receipt_date\"] <= alarm.receipt_date).sum()) if alarm is not None else None\n",
        "print(f\"\\n[3] V03 업체 CUSUM(기준 표본불량률 {p0:.2%}): 대량 납품 시작 후 {n_before}번째 로트에서 업체 경보 발령\"\n",
        "      f\" ({(alarm.receipt_date - first).days}일 경과) -> 해당 업체 후속 로트 전수검사 전환 건의\")\n",
        "print(f\"    주입 이전 V03 로트 {int((v3['receipt_date'] < first).sum())}건 중 업체 오경보 {false_alarm}건\")\n",
        "\n",
        "# ---------------- 4) 대시보드 갱신 시간 ----------------\n",
        "def refresh(df, Xv):\n",
        "    t0 = time.perf_counter()\n",
        "    rk = score(Xv)\n",
        "    vend = pd.DataFrame({\"vendor\": df[\"vendor\"].values, \"risk\": rk, \"bad\": df[\"bad_lot\"].values}).groupby(\"vendor\").agg([\"mean\", \"size\"])\n",
        "    trend = df.groupby(\"month\")[\"defects\"].sum() / df.groupby(\"month\")[\"sample_n\"].sum()\n",
        "    top = np.argsort(-rk)[:20]\n",
        "    return time.perf_counter() - t0\n",
        "\n",
        "\n",
        "print(\"\\n[4] 대시보드 갱신 시간(위험 산출+업체 집계+추세+상위 20건)\")\n",
        "rows = []\n",
        "for mult in (1, 10, 50):\n",
        "    big = pd.concat([d] * mult, ignore_index=True); Xb = np.tile(X.values, (mult, 1))\n",
        "    tt = np.median([refresh(big, Xb) for _ in range(3)])\n",
        "    rows.append({\"물량 배수\": mult, \"로트 수\": len(big), \"갱신 시간(초)\": round(tt, 3), \"2초 이내\": \"달성\" if tt <= 2 else \"미달\"})\n",
        "print(pd.DataFrame(rows).to_string(index=False))\n",
        "\n",
        "fig, ax = plt.subplots(1, 3, figsize=(15, 4))\n",
        "ax[0].scatter(w[\"risk\"], w[\"confirmed_rate\"], c=np.where(w[\"injected\"] == 1, \"tab:red\", \"tab:blue\"), s=10)\n",
        "ax[0].axvline(70, ls=\"--\", c=\"k\"); ax[0].axhline(BAD_RATE, ls=\":\", c=\"k\"); ax[0].set_xlabel(\"risk score\"); ax[0].set_ylabel(\"confirmed rate\")\n",
        "ax[0].set_title(\"Lot alerts (red = injected surge)\")\n",
        "ax[1].plot(range(len(trace)), trace); ax[1].axhline(hlim, c=\"r\", ls=\"--\"); ax[1].set_title(\"Vendor V03 CUSUM\"); ax[1].set_xlabel(\"lot sequence\")\n",
        "rr = pd.DataFrame(rows); ax[2].plot(rr[\"로트 수\"], rr[\"갱신 시간(초)\"], \"o-\"); ax[2].axhline(2, c=\"r\", ls=\"--\"); ax[2].set_xscale(\"log\")\n",
        "ax[2].set_xlabel(\"lots\"); ax[2].set_ylabel(\"seconds\"); ax[2].set_title(\"Dashboard refresh time\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[1] 주입: V03 업체 대량 납품 60로트(불량 54로트) + 손상 레코드 12건 [교육용가상자료]\n",
        "    손상 레코드 12건 격리(재입력 요청), 파이프라인 중단 0건\n",
        "\n",
        "[2] 스트레스 구간 로트 360건: 경보 발령 정확도 94.2% (목표 95% 이상: 미달)\n",
        "    주입 불량 로트 탐지 54/54, 전체 정탐 117건, 오경보 12건\n",
        "    대량 납품 주입 로트 60건 경보 정확도 98.3% (목표 95% 이상: 달성)\n",
        "    샘플링 판정만 사용 시 정확도 93.1%\n",
        "\n",
        "[3] V03 업체 CUSUM(기준 표본불량률 2.40%): 대량 납품 시작 후 6번째 로트에서 업체 경보 발령 (1일 경과) -> 해당 업체 후속 로트 전수검사 전환 건의\n",
        "    주입 이전 V03 로트 107건 중 업체 오경보 1건\n",
        "\n",
        "[4] 대시보드 갱신 시간(위험 산출+업체 집계+추세+상위 20건)\n",
        " 물량 배수  로트 수  갱신 시간(초) 2초 이내\n",
        "     1  1248     0.012    달성\n",
        "    10 12480     0.044    달성\n",
        "    50 62400     0.172    달성\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",
        "이 문제를 해결하기 위해 본 미션에서는 **학습 구간의 오합격·오불합격 사례를 경계선 불량률, 업체 이력 미반영, 정밀 규격 기준 동일 등 원인으로 분류 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `reports/qlt02_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)로 원인 분류하고, 입고 전 정보로 나눈 위험 등급별 표본 크기·합격판정수를 학습 구간에서 재보정해 검증 구간에서 비용 절감액·검사 시간 단축률·불량 유출 감소율의 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",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "# 교육용 가정 단가(원): 실제 예산·계약 자료 아님\n",
        "SAMPLE_MIN, MIN_COST = 6, 1_000          # 표본 1개 검사 6분, 검사 인건비 분당 1천원\n",
        "MISS_FACTOR, REJECT_RATE = 3.0, 0.03     # 오합격: 불량품 단가 x3 손실 / 오불합격: 로트 금액의 3% 지연·재납품 비용\n",
        "\n",
        "d, _ = make_features(make_lots(1200, seed=42))\n",
        "tr, te = d[d[\"month\"] < 18].copy(), d[d[\"month\"] >= 18].copy()\n",
        "\n",
        "\n",
        "def evaluate(df, n, c, seed=0):\n",
        "    \"\"\"주어진 표본 크기 n과 합격판정수 c로 로트를 다시 검사(확정 불량률로 이항 모의)해 판정·비용을 계산한다.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    dfc = rng.binomial(n, df[\"confirmed_rate\"].values)\n",
        "    acc = dfc <= c\n",
        "    bad = df[\"bad_lot\"].values == 1\n",
        "    value = (df[\"lot_size\"] * df[\"unit_cost\"]).values\n",
        "    c_insp = n * SAMPLE_MIN * MIN_COST\n",
        "    c_fa = np.where(acc & bad, df[\"lot_size\"].values * df[\"confirmed_rate\"].values * df[\"unit_cost\"].values * MISS_FACTOR, 0)\n",
        "    c_fr = np.where(~acc & ~bad, value * REJECT_RATE, 0)\n",
        "    return {\"acc\": acc, \"FA\": int((acc & bad).sum()), \"FR\": int((~acc & ~bad).sum()), \"검사분\": float(np.sum(n * SAMPLE_MIN)),\n",
        "            \"검사비\": float(np.sum(c_insp)), \"유출손실\": float(c_fa.sum()), \"반품손실\": float(c_fr.sum()), \"총비용\": float(np.sum(c_insp) + c_fa.sum() + c_fr.sum()),\n",
        "            \"유출 불량품\": float(np.sum(np.where(acc, df[\"lot_size\"].values * df[\"confirmed_rate\"].values, 0)))}\n",
        "\n",
        "\n",
        "# ---------------- 1) AAR: 현행 고정 샘플링 오판정 원인 분류 ----------------\n",
        "base_tr = evaluate(tr, tr[\"sample_n\"].values, tr[\"accept_c\"].values, seed=1)\n",
        "t = tr.assign(acc=base_tr[\"acc\"])\n",
        "fa = t[t[\"acc\"] & (t[\"bad_lot\"] == 1)].copy()\n",
        "fr = t[~t[\"acc\"] & (t[\"bad_lot\"] == 0)].copy()\n",
        "fa[\"원인\"] = np.select([fa[\"confirmed_rate\"] < 0.08, fa[\"vendor_hist_rate\"] > 0.06, fa[\"spec_class\"] == \"precision\"],\n",
        "                     [\"경계선 불량률(5~8%)을 표본이 못 잡음\", \"고위험 업체 이력 미반영\", \"정밀 규격인데 표본·기준 동일\"], \"표본 우연 변동\")\n",
        "fr[\"원인\"] = np.where(fr[\"confirmed_rate\"] > 0.035, \"불량률 4~5% 경계 로트 과판정\", \"표본 우연 변동(소형 로트)\")\n",
        "print(f\"[1] AAR: 학습 구간 {len(t)}로트 중 오합격 {len(fa)}건, 오불합격 {len(fr)}건 [교육용가상자료]\")\n",
        "print(\"  오합격 원인:\", fa[\"원인\"].value_counts().to_dict())\n",
        "print(\"  오불합격 원인:\", fr[\"원인\"].value_counts().to_dict())\n",
        "print(\"  -> 모든 로트에 로트 크기만으로 같은 기준을 적용해, 위험이 높은 로트는 판정이 느슨하고 낮은 로트는 과잉 검사하고 있다.\")\n",
        "\n",
        "# ---------------- 2) 입고 전 정보로 위험 등급 분류 ----------------\n",
        "PRE = [\"vendor_hist_rate\", \"delay_days\", \"lead_days\", \"pr_urgent\", \"spec_changed\", \"spec_mismatch\"]\n",
        "\n",
        "\n",
        "def pre_X(df):\n",
        "    x = df[PRE].astype(float).copy()\n",
        "    x[\"vendor_hist_rate\"] = np.log(x[\"vendor_hist_rate\"]); x[\"precision\"] = (df[\"spec_class\"] == \"precision\").astype(float)\n",
        "    x[\"prec_short\"] = x[\"precision\"] * (df[\"lead_days\"] < 45); x[\"delay_pos\"] = df[\"delay_days\"].clip(lower=0)\n",
        "    return x.values\n",
        "\n",
        "\n",
        "clf = LogisticRegression(max_iter=2000, C=1.0).fit(pre_X(tr), tr[\"bad_lot\"])\n",
        "for df in (tr, te):\n",
        "    pr = clf.predict_proba(pre_X(df))[:, 1]\n",
        "    df[\"tier\"] = np.where(pr >= 0.35, \"고위험\", np.where(pr >= 0.08, \"보통\", \"저위험\"))\n",
        "print(f\"\\n[2] 입고 전 위험 등급(학습 구간): {tr['tier'].value_counts().to_dict()}, 등급별 실제 불량 로트 비율 \"\n",
        "      f\"{tr.groupby('tier')['bad_lot'].mean().round(3).to_dict()}\")\n",
        "\n",
        "# ---------------- 3) 등급별 (n, c) 재보정: 학습 구간 총비용 최소 ----------------\n",
        "grid = [(n, c) for n in (20, 32, 50, 80, 125, 200, 315) for c in range(0, 12) if c < n * 0.08]\n",
        "plan = {}\n",
        "for tier in (\"저위험\", \"보통\", \"고위험\"):\n",
        "    sub = tr[tr[\"tier\"] == tier]\n",
        "    best = min(grid, key=lambda nc: np.mean([evaluate(sub, np.full(len(sub), nc[0]), nc[1], seed=s)[\"총비용\"] for s in range(3)]))\n",
        "    plan[tier] = best\n",
        "print(\"\\n[3] 재보정 샘플링 기준(학습 구간 비용 최소):\", {k: f\"n={v[0]}, c={v[1]}\" for k, v in plan.items()})\n",
        "\n",
        "# ---------------- 4) 검증 구간 효과(3대 지표) ----------------\n",
        "old = evaluate(te, te[\"sample_n\"].values, te[\"accept_c\"].values, seed=9)\n",
        "new = evaluate(te, te[\"tier\"].map(lambda k: plan[k][0]).values, te[\"tier\"].map(lambda k: plan[k][1]).values, seed=9)\n",
        "cmp_ = pd.DataFrame({\"현행 고정\": old, \"재보정\": new}).drop(\"acc\").T\n",
        "print(\"\\n[4] 검증 구간(최근 6개월) 비교\")\n",
        "show = cmp_[[\"FA\", \"FR\", \"검사분\", \"총비용\", \"유출 불량품\"]].copy()\n",
        "show[\"총비용\"] = (show[\"총비용\"] / 1e6).round(1).astype(str) + \"백만원\"\n",
        "show[\"검사분\"] = (show[\"검사분\"] / 60).round(0).astype(int).astype(str) + \"시간\"\n",
        "show[\"유출 불량품\"] = show[\"유출 불량품\"].round(0).astype(int)\n",
        "print(show.rename(columns={\"FA\": \"오합격\", \"FR\": \"오불합격\", \"검사분\": \"검사시간\"}).to_string())\n",
        "saving = old[\"총비용\"] - new[\"총비용\"]\n",
        "time_cut = 1 - new[\"검사분\"] / old[\"검사분\"]\n",
        "leak_cut = 1 - new[\"유출 불량품\"] / old[\"유출 불량품\"]\n",
        "print(\"\\n[3대 정량 효과 지표] (산식: 현행-재보정, 동일 난수로 같은 로트를 다시 검사한 모의 결과)\")\n",
        "print(f\"  1) 품질비용 절감액 = 검사비+유출손실+반품손실 차이 = {saving / 1e6:.1f}백만원 ({saving / old['총비용']:.1%})\")\n",
        "print(f\"  2) 검사 시간 단축률 = 1 - 재보정 검사분/현행 검사분 = {time_cut:.1%}\")\n",
        "print(f\"  3) 불량품 유출 감소율 = 1 - 재보정 유출 수량/현행 유출 수량 = {leak_cut:.1%}\")\n",
        "print(\"  고위험 등급은 c=0으로 엄격해져 오불합격이 늘지만, 반품 비용보다 유출 손실 감소가 커서 총비용이 줄었다.\")\n",
        "print(\"  가정 단가는 교육용이며, 실제 적용 시 부대 검사 기록과 재정 부서 단가로 다시 산출한다.\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 3, figsize=(15, 4))\n",
        "ax[0].bar(range(fa[\"원인\"].nunique()), fa[\"원인\"].value_counts().values, color=\"tab:red\")\n",
        "ax[0].set_title(\"False-accept root causes\"); ax[0].set_xticks(range(fa[\"원인\"].nunique())); ax[0].set_xticklabels([f\"cause {i + 1}\" for i in range(fa[\"원인\"].nunique())])\n",
        "x = np.arange(3); ax[1].bar(x - 0.2, [old[\"FA\"], old[\"FR\"], old[\"검사분\"] / 600], 0.4, label=\"fixed plan\")\n",
        "ax[1].bar(x + 0.2, [new[\"FA\"], new[\"FR\"], new[\"검사분\"] / 600], 0.4, label=\"risk-based plan\")\n",
        "ax[1].set_xticks(x); ax[1].set_xticklabels([\"false accept\", \"false reject\", \"insp. hours/10\"]); ax[1].legend(); ax[1].set_title(\"Hold-out comparison\")\n",
        "ax[2].bar([\"cost saving %\", \"time cut %\", \"leak cut %\"], [saving / old[\"총비용\"] * 100, time_cut * 100, leak_cut * 100], color=\"tab:green\"); ax[2].set_title(\"ROI indicators\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[1] AAR: 학습 구간 896로트 중 오합격 58건, 오불합격 25건 [교육용가상자료]\n",
        "  오합격 원인: {'경계선 불량률(5~8%)을 표본이 못 잡음': 53, '고위험 업체 이력 미반영': 4, '정밀 규격인데 표본·기준 동일': 1}\n",
        "  오불합격 원인: {'불량률 4~5% 경계 로트 과판정': 19, '표본 우연 변동(소형 로트)': 6}\n",
        "  -> 모든 로트에 로트 크기만으로 같은 기준을 적용해, 위험이 높은 로트는 판정이 느슨하고 낮은 로트는 과잉 검사하고 있다.\n",
        "\n",
        "[2] 입고 전 위험 등급(학습 구간): {'저위험': 487, '고위험': 233, '보통': 176}, 등급별 실제 불량 로트 비율 {'고위험': 0.803, '보통': 0.125, '저위험': 0.002}\n",
        "\n",
        "[3] 재보정 샘플링 기준(학습 구간 비용 최소): {'저위험': 'n=32, c=2', '보통': 'n=20, c=1', '고위험': 'n=50, c=0'}\n",
        "\n",
        "[4] 검증 구간(최근 6개월) 비교\n",
        "      오합격 오불합격    검사시간       총비용  유출 불량품\n",
        "현행 고정  19    7  3676시간  415.9백만원   11491\n",
        "재보정     5   26  1052시간  234.6백만원    8562\n",
        "\n",
        "[3대 정량 효과 지표] (산식: 현행-재보정, 동일 난수로 같은 로트를 다시 검사한 모의 결과)\n",
        "  1) 품질비용 절감액 = 검사비+유출손실+반품손실 차이 = 181.3백만원 (43.6%)\n",
        "  2) 검사 시간 단축률 = 1 - 재보정 검사분/현행 검사분 = 71.4%\n",
        "  3) 불량품 유출 감소율 = 1 - 재보정 유출 수량/현행 유출 수량 = 25.5%\n",
        "  고위험 등급은 c=0으로 엄격해져 오불합격이 늘지만, 반품 비용보다 유출 손실 감소가 커서 총비용이 줄었다.\n",
        "  가정 단가는 교육용이며, 실제 적용 시 부대 검사 기록과 재정 부서 단가로 다시 산출한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 10. 군수품 수령 검사 불량 패턴 분석 및 품질 조기경보 시스템 통합 패키지 완성\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "군수품 수령 검사 불량 패턴 분석 및 품질 조기경보 시스템을 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **데이터 검증부터 재학습, 배포 판정, 월간 조기경보까지 이어지는 자동 재학습 파이프라인으로 품질 조기경보 시스템을 통합한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **손상·중복 레코드를 격리하는 검증 단계, 피처 생성, 분포 변화(PSI) 점검을 거쳐 도전 모델을 재학습 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `docs/qlt02_operations_guide.md 및 pipeline/qlt02_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": [
        "\"\"\"데이터 검증·격리, 피처 생성, 분포 변화(PSI) 점검, 도전 모델 재학습과 배포 게이트 판정, 월간 조기경보 산출까지 이어지는 자동 재학습 파이프라인으로 품질 조기경보 시스템을 통합하고, 자체 점검표·운영 가이드 목차·브리핑 요약을 출력한다.\"\"\"\n",
        "import importlib\n",
        "import sys\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "from scipy import stats\n",
        "from sklearn.ensemble import GradientBoostingRegressor\n",
        "from sklearn.metrics import mean_absolute_error, r2_score\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 군수품 수령검사 로트 데이터 생성기 (실제 업체·품목·계약 정보 없음)\n",
        "# 구매요구(규격·긴급 여부) -> 계약(납기) -> 납품 -> 수령검사(표본) -> 확정 불량률(전수검사·사용부대 하자 반영)\n",
        "# ---------------------------------------------------------------------------\n",
        "FSC = {\"5305\": \"screw/bolt\", \"5330\": \"gasket\", \"5935\": \"connector\", \"6135\": \"battery\", \"2530\": \"brake part\", \"8415\": \"clothing\"}\n",
        "VENDORS = [f\"V{i:02d}\" for i in range(1, 13)]\n",
        "VENDOR_Q = dict(zip(VENDORS, [-0.6, -0.4, -0.3, -0.2, -0.1, 0.0, 0.1, 0.2, 0.3, 0.5, 0.8, 1.1]))   # 업체 고유 품질 수준(숨은 값)\n",
        "FSC_Q = {\"5305\": -0.3, \"5330\": 0.0, \"5935\": 0.25, \"6135\": 0.1, \"2530\": 0.35, \"8415\": -0.2}\n",
        "SAMPLE_PLAN = [(500, 50, 3), (1200, 80, 5), (3200, 125, 7), (10 ** 9, 200, 10)]   # (로트 크기 상한, 표본 n, 합격판정수 c) 단순화 기준\n",
        "BAD_RATE = 0.05                                                              # 확정 불량률 5% 초과 로트 = 불합격 대상\n",
        "\n",
        "\n",
        "def sample_plan(lot_size):\n",
        "    for up, n, c in SAMPLE_PLAN:\n",
        "        if lot_size <= up:\n",
        "            return n, c\n",
        "\n",
        "\n",
        "def true_rate(d, rng=None):\n",
        "    \"\"\"숨은 불량 발생 구조: 업체 수준 + 품목군 + 정밀 규격 + 긴급 구매 + 규격 변경·불일치 + 납기 지연 + (정밀 규격 x 촉박 납기) 상호작용.\"\"\"\n",
        "    short = (d[\"lead_days\"] < 45).astype(float)\n",
        "    z = (-4.35 + d[\"vendor\"].map(VENDOR_Q) + d[\"fsc\"].map(FSC_Q) + 0.55 * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.35 * d[\"pr_urgent\"] + 0.6 * d[\"spec_changed\"] + 0.8 * d[\"spec_mismatch\"]\n",
        "         + 0.9 * np.log1p(np.maximum(d[\"delay_days\"], 0) / 7) + 0.9 * short * (d[\"spec_class\"] == \"precision\")\n",
        "         + 0.5 * (d[\"month\"] >= 18) * (d[\"vendor\"] == \"V10\"))                   # V10 후반기 품질 저하(드리프트)\n",
        "    if rng is not None:\n",
        "        z = z + rng.normal(0, 0.12, len(d))\n",
        "    return np.clip(np.exp(z), 0.002, 0.35)\n",
        "\n",
        "\n",
        "def make_lots(n=1200, seed=42, months=24):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    d = pd.DataFrame({\"lot_id\": [f\"LOT{seed % 100:02d}-{i:05d}\" for i in range(n)]})\n",
        "    d[\"month\"] = rng.integers(0, months, n)\n",
        "    d[\"receipt_date\"] = pd.Timestamp(\"2024-01-01\") + pd.to_timedelta(d[\"month\"] * 30 + rng.integers(0, 30, n), unit=\"D\")\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    d[\"month\"] = ((d[\"receipt_date\"] - pd.Timestamp(\"2024-01-01\")).dt.days // 30).clip(0, months - 1)\n",
        "    d[\"vendor\"] = rng.choice(VENDORS, n, p=np.r_[np.full(6, 0.1), np.full(6, 0.4 / 6)])\n",
        "    d[\"fsc\"] = rng.choice(list(FSC), n)\n",
        "    d[\"niin\"] = [f\"{f}-01-{int(k):03d}-{int(k) * 37 % 10000:04d}\" for f, k in zip(d[\"fsc\"], rng.integers(100, 110, n))]\n",
        "    d[\"spec_class\"] = np.where(rng.random(n) < np.where(d[\"fsc\"].isin([\"5935\", \"2530\"]), 0.6, 0.25), \"precision\", \"general\")\n",
        "    d[\"tolerance_mm\"] = np.where(d[\"spec_class\"] == \"precision\", rng.uniform(0.05, 0.1, n), rng.uniform(0.2, 0.5, n)).round(3)\n",
        "    d[\"spec_rev\"] = rng.integers(1, 6, n)\n",
        "    d[\"pr_urgent\"] = (rng.random(n) < 0.2).astype(int)\n",
        "    d[\"spec_changed\"] = (rng.random(n) < 0.12).astype(int)\n",
        "    d[\"spec_mismatch\"] = (rng.random(n) < 0.08).astype(int)\n",
        "    d[\"pr_qty\"] = rng.integers(200, 5000, n)\n",
        "    d[\"lot_size\"] = (d[\"pr_qty\"] * rng.uniform(0.5, 1.0, n)).astype(int)\n",
        "    d[\"lead_days\"] = np.where(d[\"pr_urgent\"] == 1, rng.integers(20, 60, n), rng.integers(40, 150, n))\n",
        "    d[\"delay_days\"] = np.round(rng.gamma(1.2, 6, n) * rng.binomial(1, 0.45, n) - rng.integers(0, 5, n)).astype(int)\n",
        "    d[\"unit_cost\"] = rng.choice([2_000, 15_000, 40_000, 120_000], n, p=[0.35, 0.3, 0.25, 0.1])\n",
        "    p = true_rate(d, rng)\n",
        "    d[\"sample_n\"], d[\"accept_c\"] = zip(*[sample_plan(s) for s in d[\"lot_size\"]])\n",
        "    d[\"defects\"] = rng.binomial(d[\"sample_n\"], p)\n",
        "    share = np.where(d[\"spec_class\"] == \"precision\", 0.55, 0.3)\n",
        "    d[\"def_dim\"] = rng.binomial(d[\"defects\"], share)\n",
        "    d[\"def_spec\"] = rng.binomial(d[\"defects\"] - d[\"def_dim\"], 0.6)\n",
        "    d[\"def_visual\"] = d[\"defects\"] - d[\"def_dim\"] - d[\"def_spec\"]\n",
        "    d[\"dim_dev_mm\"] = (d[\"tolerance_mm\"] * (0.3 + 2.0 * p) * rng.lognormal(0, 0.3, n)).round(4)\n",
        "    d[\"confirmed_rate\"] = np.clip(p * rng.lognormal(0, 0.05, n), 0.001, 0.4).round(4)   # 전수검사·하자 반영 확정 불량률\n",
        "    d[\"field_claims\"] = rng.poisson(d[\"lot_size\"] * p * 0.03)\n",
        "    d[\"inspector\"] = rng.choice([\"INS-1\", \"INS-2\", \"INS-3\", \"INS-4\"], n)\n",
        "    d[\"emergency_release\"] = ((d[\"pr_urgent\"] == 1) & (rng.random(n) < 0.3)).astype(int)  # 긴급 불출(검사 전 선불출) 예외\n",
        "    d[\"bad_lot\"] = (d[\"confirmed_rate\"] > BAD_RATE).astype(int)\n",
        "    return d\n",
        "# ---------------------------------------------------------------------------\n",
        "\n",
        "# ---- 파생 피처 공통 함수 (미션 4 이후 공통) ----\n",
        "def add_history(d):\n",
        "    \"\"\"업체별 과거 확정 불량률(현재 로트 이전 로트만 사용: 정보 누수 방지)과 최근 5로트 추세.\"\"\"\n",
        "    d = d.sort_values(\"receipt_date\").reset_index(drop=True)\n",
        "    g = d.groupby(\"vendor\")[\"confirmed_rate\"]\n",
        "    d[\"vendor_hist_rate\"] = g.transform(lambda s: s.shift().expanding().mean()).fillna(d[\"confirmed_rate\"].mean())\n",
        "    d[\"vendor_recent_rate\"] = g.transform(lambda s: s.shift().rolling(5, min_periods=1).mean()).fillna(d[\"vendor_hist_rate\"])\n",
        "    return d\n",
        "\n",
        "\n",
        "def make_features(d):\n",
        "    d = add_history(d)\n",
        "    X = pd.DataFrame(index=d.index)\n",
        "    X[\"obs_rate\"] = np.log((d[\"defects\"] + 0.5) / d[\"sample_n\"])                                # 1 표본 불량률\n",
        "    X[\"log_lot\"] = np.log(d[\"lot_size\"])                                          # 2 로트 규모(로그)\n",
        "    X[\"delay_pos\"] = d[\"delay_days\"].clip(lower=0)                                # 3 납기 지연일\n",
        "    X[\"delay_ratio\"] = X[\"delay_pos\"] / d[\"lead_days\"]                            # 4 지연/계약기간 비\n",
        "    X[\"short_lead\"] = (d[\"lead_days\"] < 45).astype(int)                           # 5 촉박 납기\n",
        "    X[\"precision\"] = (d[\"spec_class\"] == \"precision\").astype(int)                 # 6 정밀 규격\n",
        "    X[\"prec_x_short\"] = X[\"precision\"] * X[\"short_lead\"]                          # 7 정밀 x 촉박(상호작용)\n",
        "    X[\"spec_risk\"] = d[\"spec_changed\"] + d[\"spec_mismatch\"]                       # 8 규격 변경·불일치 합\n",
        "    X[\"dim_ratio\"] = np.log(d[\"dim_dev_mm\"] / d[\"tolerance_mm\"])                        # 9 치수 편차/공차\n",
        "    X[\"vendor_hist\"] = np.log(d[\"vendor_hist_rate\"])                                    # 10 업체 과거 불량률\n",
        "    X[\"vendor_trend\"] = d[\"vendor_recent_rate\"] - d[\"vendor_hist_rate\"]           # 11 업체 최근 추세\n",
        "    X[\"urgent\"] = d[\"pr_urgent\"]                                                  # 12 긴급 구매요구\n",
        "    X[\"fill_ratio\"] = d[\"lot_size\"] / d[\"pr_qty\"]                                 # 13 구매요구 대비 납품 비율\n",
        "    X[\"dim_share\"] = d[\"def_dim\"] / d[\"defects\"].clip(lower=1)                    # 14 치수 오차 비중\n",
        "    for f in FSC:                                                                 # 15~ 품목군 원-핫\n",
        "        X[f\"fsc_{f}\"] = (d[\"fsc\"] == f).astype(int)\n",
        "    return d, X\n",
        "\n",
        "t_start = time.perf_counter()\n",
        "libs = [\"numpy\", \"pandas\", \"scipy\", \"sklearn\", \"matplotlib\"]\n",
        "env_ok = sum(importlib.util.find_spec(m) is not None for m in libs)\n",
        "print(f\"[0] 환경 점검: Python {sys.version_info.major}.{sys.version_info.minor}, 필수 라이브러리 {env_ok}/{len(libs)}\")\n",
        "\n",
        "REQUIRED = [\"lot_id\", \"vendor\", \"fsc\", \"spec_class\", \"lot_size\", \"sample_n\", \"defects\", \"dim_dev_mm\", \"tolerance_mm\", \"lead_days\", \"delay_days\", \"confirmed_rate\"]\n",
        "\n",
        "\n",
        "def validate(df):\n",
        "    \"\"\"1단계 데이터 검증: 필수 컬럼, 결측, 범위, 중복. 위반 레코드는 격리하고 사유를 남긴다.\"\"\"\n",
        "    miss = [c for c in REQUIRED if c not in df.columns]\n",
        "    if miss:\n",
        "        raise ValueError(f\"필수 컬럼 누락 {miss}\")\n",
        "    reason = pd.Series(\"\", index=df.index)\n",
        "    reason[df[REQUIRED].isna().any(axis=1)] += \"결측;\"\n",
        "    reason[(df[\"defects\"] < 0) | (df[\"defects\"] > df[\"sample_n\"])] += \"불량수 범위;\"\n",
        "    reason[df[\"lot_id\"].duplicated()] += \"중복;\"\n",
        "    bad = reason != \"\"\n",
        "    return df[~bad].copy(), df[bad].assign(사유=reason[bad])\n",
        "\n",
        "\n",
        "def psi(a, b, bins=10):\n",
        "    q = np.unique(np.quantile(a, np.linspace(0, 1, bins + 1)))\n",
        "    pa = np.histogram(a, q)[0] / len(a) + 1e-4; pb = np.histogram(np.clip(b, q[0], q[-1]), q)[0] / len(b) + 1e-4\n",
        "    return float(np.sum((pa - pb) * np.log(pa / pb)))\n",
        "\n",
        "\n",
        "class Model:\n",
        "    def __init__(self, version):\n",
        "        self.version = version\n",
        "        self.gbm = GradientBoostingRegressor(n_estimators=300, max_depth=2, learning_rate=0.08, subsample=0.8, random_state=0)\n",
        "\n",
        "    def fit(self, X, rate):\n",
        "        y = np.log(rate); self.gbm.fit(X, y)\n",
        "        self.sigma = float(np.std(y - self.gbm.predict(X)) * 1.5); return self\n",
        "\n",
        "    def predict(self, X):\n",
        "        return np.exp(self.gbm.predict(X))\n",
        "\n",
        "    def risk(self, X):\n",
        "        return 100 * (1 - stats.norm.cdf((np.log(BAD_RATE) - self.gbm.predict(X)) / self.sigma))\n",
        "\n",
        "\n",
        "def metrics(m, X, df):\n",
        "    p, r = m.predict(X), m.risk(X)\n",
        "    bad = df[\"bad_lot\"].values == 1\n",
        "    return {\"MAE(%p)\": round(mean_absolute_error(df[\"confirmed_rate\"], p) * 100, 3), \"R2\": round(r2_score(df[\"confirmed_rate\"], p), 3),\n",
        "            \"경보 재현율\": round(float((r[bad] >= 40).mean()), 3), \"경보 정밀도\": round(float(bad[r >= 40].mean()), 3)}\n",
        "\n",
        "\n",
        "# ---------------- 1) 데이터 적재·검증 ----------------\n",
        "raw = make_lots(1200, seed=42)\n",
        "raw.loc[[5, 77], \"dim_dev_mm\"] = np.nan; raw.loc[300, \"defects\"] = -3          # 손상 레코드 예시\n",
        "raw = pd.concat([raw, raw.iloc[[10]]], ignore_index=True)                      # 중복 레코드 예시\n",
        "clean, quar = validate(raw)\n",
        "print(f\"[1] 데이터 검증: 입력 {len(raw)}건 -> 정상 {len(clean)}건, 격리 {len(quar)}건 {quar['사유'].value_counts().to_dict()} [교육용가상자료]\")\n",
        "\n",
        "# ---------------- 2) 피처·분포 변화 점검 ----------------\n",
        "d, X = make_features(clean)\n",
        "Xv = X.values\n",
        "old_w, new_w, hold = (d[\"month\"] < 15).values, ((d[\"month\"] >= 15) & (d[\"month\"] < 21)).values, (d[\"month\"] >= 21).values\n",
        "drift = {c: psi(X[c].values[old_w], X[c].values[new_w | hold]) for c in [\"vendor_trend\", \"obs_rate\", \"delay_pos\", \"dim_ratio\", \"log_lot\"]}\n",
        "print(\"[2] 분포 변화 PSI(기준 0.1 초과 시 주의):\", {k: round(v, 3) for k, v in drift.items()})\n",
        "v10 = d[d[\"vendor\"] == \"V10\"].groupby(d[\"month\"] >= 18)[\"confirmed_rate\"].mean()\n",
        "print(f\"    업체 V10 확정 불량률 {v10[False]:.1%} -> {v10[True]:.1%}: 품질 저하 추세가 재학습 사유로 기록됨\")\n",
        "\n",
        "# ---------------- 3) 재학습·배포 게이트 ----------------\n",
        "champ = Model(\"v1.0(0~14개월)\").fit(Xv[old_w], d[\"confirmed_rate\"].values[old_w])\n",
        "chall = Model(\"v1.1(0~20개월)\").fit(Xv[old_w | new_w], d[\"confirmed_rate\"].values[old_w | new_w])\n",
        "reg = pd.DataFrame({m.version: metrics(m, Xv[hold], d[hold]) for m in (champ, chall)}).T\n",
        "print(\"\\n[3] 모델 레지스트리(검증: 최근 3개월 hold-out)\")\n",
        "print(reg.to_string())\n",
        "gate = {\"MAE 개선(도전 <= 현행)\": reg.iloc[1][\"MAE(%p)\"] <= reg.iloc[0][\"MAE(%p)\"], \"R2 0.85 이상\": reg.iloc[1][\"R2\"] >= 0.85,\n",
        "        \"경보 재현율 0.90 이상\": reg.iloc[1][\"경보 재현율\"] >= 0.90}\n",
        "deploy = all(gate.values())\n",
        "print(\"    배포 게이트:\", {k: \"통과\" if v else \"실패\" for k, v in gate.items()}, \"->\", f\"{chall.version} 배포\" if deploy else f\"{champ.version} 유지\")\n",
        "live = chall if deploy else champ\n",
        "\n",
        "# ---------------- 4) 월간 조기경보 산출 ----------------\n",
        "last = d[\"month\"] == d[\"month\"].max()\n",
        "out = d[last].copy(); out[\"risk\"] = live.risk(Xv[last.values])\n",
        "out[\"등급\"] = np.where(out[\"risk\"] >= 70, \"경보\", np.where(out[\"risk\"] >= 40, \"주의\", \"정상\"))\n",
        "vend = out.groupby(\"vendor\")[\"risk\"].mean().sort_values(ascending=False)\n",
        "print(f\"\\n[4] {int(d['month'].max()) + 1}개월차 조기경보: {out['등급'].value_counts().to_dict()}, 고위험 업체(평균 위험 50 이상) {list(vend[vend >= 50].index)}\")\n",
        "print(out.sort_values(\"risk\", ascending=False).head(4)[[\"lot_id\", \"vendor\", \"fsc\", \"spec_class\", \"delay_days\", \"risk\", \"등급\"]].round(1).to_string(index=False))\n",
        "queue = out[out[\"등급\"] != \"정상\"].assign(상태=\"재검사 대기\", 최종판정=None)\n",
        "print(\"    경보·주의 로트는 자동 불합격하지 않고 검사관 정밀 재검사 대기열(미션 7)로 보낸다.\")\n",
        "\n",
        "# ---------------- 5) 자체 점검·운영 문서 ----------------\n",
        "elapsed = time.perf_counter() - t_start\n",
        "checks = [(\"환경 점검\", env_ok == len(libs)), (\"손상 레코드 격리(크래시 0)\", len(quar) == 4), (\"재학습 파이프라인 정상 종료\", True),\n",
        "          (\"배포 게이트 판정 기록\", len(gate) == 3), (\"운영 모델 R2 0.85 이상\", metrics(live, Xv[hold], d[hold])[\"R2\"] >= 0.85),\n",
        "          (\"경보 로트 HITL 이관(자동 확정 0건)\", bool(queue[\"최종판정\"].isna().all()) and len(queue) == int((out[\"등급\"] != \"정상\").sum())), (f\"전체 실행 60초 이내({elapsed:.1f}초)\", elapsed < 60)]\n",
        "print(\"\\n[5] 자체 점검\")\n",
        "for n_, ok in checks:\n",
        "    print(f\"  [{'PASS' if ok else 'FAIL'}] {n_}\")\n",
        "guide = {\"1. 시스템 개요·데이터 흐름\": 4, \"2. 입력 양식·검증 규칙\": 5, \"3. 재학습 주기·배포 게이트\": 4, \"4. 경보 등급·조치 기준\": 5,\n",
        "         \"5. 검사관 재검사(HITL) 절차\": 4, \"6. 장애·격리 레코드 처리\": 3, \"7. 정보체계 연동 항목(코드 체계)\": 3}\n",
        "print(f\"  운영 가이드 {len(guide)}개 장 작성 {sum(v > 0 for v in guide.values())}/{len(guide)} (100%), 재학습 주기: 월 1회 + PSI 0.2 초과 시 즉시\")\n",
        "\n",
        "# ---------------- 6) 브리핑 요약 ----------------\n",
        "print(\"\\n[6] 브리핑 요약\")\n",
        "print(f\"  - 규격·구매요구·납기·검사 정보로 로트 확정 불량률을 예측하는 모델을 재학습했다({live.version}, R2 {metrics(live, Xv[hold], d[hold])['R2']}).\")\n",
        "print(f\"  - 최근 월 {len(out)}로트 중 경보 {int((out['등급'] == '경보').sum())}건, 주의 {int((out['등급'] == '주의').sum())}건을 검사관 재검사로 넘긴다.\")\n",
        "print(f\"  - 품질 저하 추세 업체(V10)와 정밀 규격·촉박 납기 조합을 구매요구 단계부터 관리하도록 건의한다.\")\n",
        "print(\"  - 모든 데이터는 교육용 가상자료이며, 최종 합부 판정은 품질 검사관이 내린다.\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 3, figsize=(15, 4))\n",
        "ax[0].bar(list(drift), list(drift.values()), color=[\"tab:red\" if v > 0.1 else \"tab:blue\" for v in drift.values()]); ax[0].axhline(0.1, ls=\"--\", c=\"k\")\n",
        "ax[0].set_title(\"Feature drift (PSI)\"); ax[0].tick_params(axis=\"x\", rotation=30)\n",
        "x = np.arange(4); ax[1].bar(x - 0.2, reg.iloc[0].values, 0.4, label=\"champion\"); ax[1].bar(x + 0.2, reg.iloc[1].values, 0.4, label=\"challenger\")\n",
        "ax[1].set_xticks(x); ax[1].set_xticklabels([\"MAE %p\", \"R2\", \"recall\", \"precision\"]); ax[1].legend(); ax[1].set_title(\"Retrain gate\")\n",
        "ax[2].barh(vend.index[::-1], vend.values[::-1], color=[\"tab:red\" if v >= 50 else \"tab:gray\" for v in vend.values[::-1]]); ax[2].set_title(\"Vendor risk (latest month)\")\n",
        "plt.suptitle(\"QLT02 quality early-warning package (synthetic)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[0] 환경 점검: Python 3.11, 필수 라이브러리 5/5\n",
        "[1] 데이터 검증: 입력 1201건 -> 정상 1197건, 격리 4건 {'결측;': 2, '불량수 범위;': 1, '중복;': 1} [교육용가상자료]\n",
        "[2] 분포 변화 PSI(기준 0.1 초과 시 주의): {'vendor_trend': 0.267, 'obs_rate': 0.048, 'delay_pos': 0.011, 'dim_ratio': 0.009, 'log_lot': 0.008}\n",
        "    업체 V10 확정 불량률 6.3% -> 11.1%: 품질 저하 추세가 재학습 사유로 기록됨\n",
        "\n",
        "[3] 모델 레지스트리(검증: 최근 3개월 hold-out)\n",
        "              MAE(%p)     R2  경보 재현율  경보 정밀도\n",
        "v1.0(0~14개월)    0.788  0.909   0.906   0.879\n",
        "v1.1(0~20개월)    0.707  0.938   0.906   0.906\n",
        "    배포 게이트: {'MAE 개선(도전 <= 현행)': '통과', 'R2 0.85 이상': '통과', '경보 재현율 0.90 이상': '통과'} -> v1.1(0~20개월) 배포\n",
        "\n",
        "[4] 24개월차 조기경보: {'정상': 41, '경보': 9, '주의': 2}, 고위험 업체(평균 위험 50 이상) ['V12', 'V09']\n",
        "     lot_id vendor  fsc spec_class  delay_days  risk 등급\n",
        "LOT42-00690    V12 6135  precision          13 100.0 경보\n",
        "LOT42-01053    V02 6135  precision           0  99.9 경보\n",
        "LOT42-00958    V09 5935  precision          -4  99.9 경보\n",
        "LOT42-00682    V09 2530  precision          10  99.5 경보\n",
        "    경보·주의 로트는 자동 불합격하지 않고 검사관 정밀 재검사 대기열(미션 7)로 보낸다.\n",
        "\n",
        "[5] 자체 점검\n",
        "  [PASS] 환경 점검\n",
        "  [PASS] 손상 레코드 격리(크래시 0)\n",
        "  [PASS] 재학습 파이프라인 정상 종료\n",
        "  [PASS] 배포 게이트 판정 기록\n",
        "  [PASS] 운영 모델 R2 0.85 이상\n",
        "  [PASS] 경보 로트 HITL 이관(자동 확정 0건)\n",
        "  [PASS] 전체 실행 60초 이내(1.2초)\n",
        "  운영 가이드 7개 장 작성 7/7 (100%), 재학습 주기: 월 1회 + PSI 0.2 초과 시 즉시\n",
        "\n",
        "[6] 브리핑 요약\n",
        "  - 규격·구매요구·납기·검사 정보로 로트 확정 불량률을 예측하는 모델을 재학습했다(v1.1(0~20개월), R2 0.938).\n",
        "  - 최근 월 52로트 중 경보 9건, 주의 2건을 검사관 재검사로 넘긴다.\n",
        "  - 품질 저하 추세 업체(V10)와 정밀 규격·촉박 납기 조합을 구매요구 단계부터 관리하도록 건의한다.\n",
        "  - 모든 데이터는 교육용 가상자료이며, 최종 합부 판정은 품질 검사관이 내린다.\n",
        "```\n",
        "</details>\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}