{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# 🎯 [ADM01] 결원·근무부하 추세 대시보드와 보충 필요량 예측\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/ADM01/ADM01_solution.ipynb)\n",
        "\n",
        "---\n",
        "\n",
        "## 📌 1. 프로젝트 개요 및 배경\n",
        "\n",
        "| 항목 | 내용 |\n",
        "| :--- | :--- |\n",
        "| **과정 분류** | 행정·인사 (Level 3) |\n",
        "| **학습 소요 시간** | 10개 미션 (5일 / 총 27.5시간) |\n",
        "| **환경 사양** | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |\n",
        "\n",
        "### 📖 프로젝트 목표\n",
        "병력 감소와 전출·파견·휴가·환자·교육이 겹치면 부대 직무별 결원이 갑자기 커지고 남은 인원의 초과근무와 피로가 누적되어 전투력 유지가 어려워지지만, 현재원 표만으로는 향후 결원과 보충 필요량을 미리 보기 어렵다. 본 프로젝트는 러시아-우크라이나 전쟁에서 드러난 인력 소모·피로 관리 문제의식을 교육용 가상 자료로 재구성해, 가상 대대 4곳의 직무 6종 주간 인력 집계 자료로 결원·근무부하 추세를 보고 향후 3개월 보충 필요량을 예측하는 과제이다. 인력 자료 스키마와 근무일지 전처리, 피로도 Baseline, 13주 결원 예측과 위험 지수·신뢰도, 현황 대시보드, 인사관·지휘관 검토 워크플로우, 대량 결원 스트레스 시험, 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",
        "부대 인사 데이터(현재원, 결원, 휴가, 환자, 교대주기) 스키마를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **가상 부대의 직무별 인력 현황(편제·현재원·결원·휴가·환자·교육·보충·전출·초과근무·교대주기) 자료의 구조를 JSON 스키마로 정의하고, 정상 표본 검증 오류 0건과 필드 정의 완성도 100%를 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **가상 대대 4곳과 직무 6종의 주간 집계 자료에서 필드 15개의 자료형·단위·허용 범위·설명·출처를 모두 정의하고 필드 간 논리 규칙(결원은 편제에서 현재원을 뺀 값, 휴가·환자·교육 합계는 현재원 이하) 작성하여 원인을 분석하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `data/schema/adm01_input_schema.json 및 docs/adm01_data_spec.md` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 행정·인사 실무 및 러시아-우크라이나 전쟁 전장 환경과의 군사적 부합성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 스키마 필드 정의 완성도 100%, 샘플 데이터 검증 오류 0건\n",
        "\n",
        "\n",
        "- 진행 기준: 입력 데이터 유효성 검증 스크립트 통과 및 명세서 문서화 완료\n",
        "\n",
        "\n",
        "- 검증 방법: JSON Schema 검증 도구 및 샘플 데이터 파싱 단위 테스트\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"가상 부대 인력 현황(현재원·결원·휴가·환자·교육·보충·전출·초과근무·교대주기, 15개 필드)의 JSON 스키마와 검증 규칙을 정의하고, 정상 표본의 검증 오류 0건·스키마 필드 정의 완성도 100%를 확인하며 형식·범위·논리 규칙 위반 오류를 주입해 검출력과 개인 식별 정보 부재를 점검한다.\"\"\"\n",
        "import json\n",
        "import re\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# 1) 스키마: 필드마다 자료형·단위·허용 범위·설명·출처를 모두 정의한다(완성도 점검 대상)\n",
        "SCHEMA = {\n",
        "    \"unit\": {\"type\": \"string\", \"enum\": UNITS, \"unit\": \"-\", \"desc\": \"소속 가상 대대\", \"source\": \"인사 편제 집계\"},\n",
        "    \"job\": {\"type\": \"string\", \"enum\": JOBS, \"unit\": \"-\", \"desc\": \"직무 구분(가상)\", \"source\": \"인사 편제 집계\"},\n",
        "    \"week\": {\"type\": \"integer\", \"min\": 1, \"max\": 156, \"unit\": \"주\", \"desc\": \"기준 주차\", \"source\": \"주간 인사 보고\"},\n",
        "    \"authorized\": {\"type\": \"integer\", \"min\": 1, \"max\": 200, \"unit\": \"명\", \"desc\": \"편제 인원\", \"source\": \"인사 편제 집계\"},\n",
        "    \"present\": {\"type\": \"integer\", \"min\": 0, \"max\": 200, \"unit\": \"명\", \"desc\": \"현재원\", \"source\": \"주간 인사 보고\"},\n",
        "    \"vacancy\": {\"type\": \"integer\", \"min\": 0, \"max\": 200, \"unit\": \"명\", \"desc\": \"결원 = 편제 - 현재원\", \"source\": \"계산\"},\n",
        "    \"leave\": {\"type\": \"integer\", \"min\": 0, \"max\": 200, \"unit\": \"명\", \"desc\": \"휴가 인원\", \"source\": \"휴가 현황\"},\n",
        "    \"sick\": {\"type\": \"integer\", \"min\": 0, \"max\": 200, \"unit\": \"명\", \"desc\": \"환자(진료·입원) 인원, 개인 식별 없이 수만 집계\", \"source\": \"의무 집계\"},\n",
        "    \"training\": {\"type\": \"integer\", \"min\": 0, \"max\": 200, \"unit\": \"명\", \"desc\": \"교육·파견 인원\", \"source\": \"교육 계획\"},\n",
        "    \"inflow\": {\"type\": \"integer\", \"min\": 0, \"max\": 100, \"unit\": \"명\", \"desc\": \"주간 보충(전입) 인원\", \"source\": \"인사 명령 집계\"},\n",
        "    \"outflow\": {\"type\": \"integer\", \"min\": 0, \"max\": 100, \"unit\": \"명\", \"desc\": \"주간 전출·전역 인원\", \"source\": \"인사 명령 집계\"},\n",
        "    \"overtime_h\": {\"type\": \"number\", \"min\": 0, \"max\": 80, \"unit\": \"시간/인/주\", \"desc\": \"가용 인원 1명당 주간 초과근무 시간\", \"source\": \"근무일지 집계\"},\n",
        "    \"night_ratio\": {\"type\": \"number\", \"min\": 0, \"max\": 1, \"unit\": \"비율\", \"desc\": \"야간 근무 비율\", \"source\": \"근무일지 집계\"},\n",
        "    \"rotation_days\": {\"type\": \"integer\", \"min\": 7, \"max\": 365, \"unit\": \"일\", \"desc\": \"교대·보충 주기\", \"source\": \"인사 계획\"},\n",
        "    \"fatigue_label\": {\"type\": \"integer\", \"enum\": [0, 1], \"unit\": \"-\", \"desc\": \"고피로 여부(검증용 정답 라벨, 가상)\", \"source\": \"가상 평가 기준\"},\n",
        "}\n",
        "print(f\"[스키마] 필드 {len(SCHEMA)}개, 필수 속성(type·unit·desc·source) 모두 정의된 필드 {sum(all(k in v for k in ('type', 'unit', 'desc', 'source')) for v in SCHEMA.values())}개\")\n",
        "assert list(SCHEMA) == FIELDS\n",
        "PII = re.compile(r\"name|이름|군번|주민|rrn|phone|전화|address|주소\", re.I)\n",
        "print(f\"개인 식별 필드 {sum(bool(PII.search(k + v['desc'])) for k, v in SCHEMA.items())}개(집계 인원 수만 사용)\")\n",
        "\n",
        "\n",
        "def check(rec):\n",
        "    errs = []\n",
        "    for k, r in SCHEMA.items():\n",
        "        if k not in rec or rec[k] is None or (isinstance(rec[k], float) and np.isnan(rec[k])):\n",
        "            errs.append(f\"{k}: 필수 값 누락\"); continue\n",
        "        v = rec[k]\n",
        "        ok_t = {\"string\": isinstance(v, str), \"integer\": isinstance(v, (int, np.integer)) and not isinstance(v, bool), \"number\": isinstance(v, (int, float, np.number)) and not isinstance(v, bool)}[r[\"type\"]]\n",
        "        if not ok_t:\n",
        "            errs.append(f\"{k}: 자료형 오류({type(v).__name__})\"); continue\n",
        "        if \"enum\" in r and v not in r[\"enum\"]:\n",
        "            errs.append(f\"{k}: 허용되지 않은 값 '{v}'\")\n",
        "        if \"min\" in r and v < r[\"min\"]:\n",
        "            errs.append(f\"{k}: 최소값 {r['min']} 미만\")\n",
        "        if \"max\" in r and v > r[\"max\"]:\n",
        "            errs.append(f\"{k}: 최대값 {r['max']} 초과\")\n",
        "    if not errs:                                                                    # 필드 간 논리 규칙\n",
        "        if rec[\"vacancy\"] != rec[\"authorized\"] - rec[\"present\"]:\n",
        "            errs.append(\"vacancy: 편제 - 현재원과 불일치\")\n",
        "        if rec[\"leave\"] + rec[\"sick\"] + rec[\"training\"] > rec[\"present\"]:\n",
        "            errs.append(\"leave+sick+training: 현재원 초과\")\n",
        "        if rec[\"present\"] > rec[\"authorized\"]:\n",
        "            errs.append(\"present: 편제 인원 초과\")\n",
        "    return errs\n",
        "\n",
        "\n",
        "df = make_manning(7, 104)\n",
        "sample = df[df[\"week\"].isin([10, 50])].to_dict(\"records\")                          # 24개 셀 x 2주 = 48건 + 전체 점검\n",
        "recs = df.to_dict(\"records\")\n",
        "bad_all = sum(bool(check(r)) for r in recs)\n",
        "bad_sample = sum(bool(check(r)) for r in sample)\n",
        "print(f\"[정상 자료 검증] 표본 {len(sample)}건 오류 {bad_sample}건, 전체 {len(recs)}건 오류 {bad_all}건 (목표 오류 0건: {'달성' if bad_all == 0 else '미달'})\")\n",
        "\n",
        "# 2) 오류 주입 시험\n",
        "r0 = recs[100]\n",
        "inj = {\"자료형 오류\": dict(r0, authorized=\"서른\"), \"허용값 오류\": dict(r0, job=\"미상직무\"), \"범위 오류(음수 초과근무)\": dict(r0, overtime_h=-4.0), \"논리 오류(결원 불일치)\": dict(r0, vacancy=r0[\"vacancy\"] + 3),\n",
        "       \"논리 오류(휴가+환자+교육>현재원)\": dict(r0, leave=r0[\"present\"] + 1), \"필드 누락\": {k: v for k, v in r0.items() if k != \"night_ratio\"}, \"결측값\": dict(r0, present=np.nan),\n",
        "       \"비율 범위 초과\": dict(r0, night_ratio=1.4)}\n",
        "print(\"\\n[오류 주입 시험]\")\n",
        "caught = 0\n",
        "for name, rec in inj.items():\n",
        "    e = check(rec); caught += bool(e); print(f\"  {name:22s}-> {'검출' if e else '미검출'}: {e[0] if e else ''}\")\n",
        "print(f\"주입 오류 {len(inj)}종 중 {caught}종 검출\")\n",
        "\n",
        "# 3) 자료 구조 요약과 입출력 명세\n",
        "print(\"\\n[자료 구조 요약] 셀(부대 x 직무) %d개 x 104주 = %d행\" % (df.groupby([\"unit\", \"job\"]).ngroups, len(df)))\n",
        "s = df.assign(vr=df[\"vacancy\"] / df[\"authorized\"]).groupby(\"job\").agg(평균편제=(\"authorized\", \"mean\"), 결원율=(\"vr\", \"mean\"), 평균초과근무=(\"overtime_h\", \"mean\"), 고피로비율=(\"fatigue_label\", \"mean\"))\n",
        "print(s.round(2).to_string())\n",
        "js = json.dumps({\"title\": \"adm01_input_schema\", \"fields\": {k: {kk: vv for kk, vv in v.items() if kk != \"desc\"} for k, v in list(SCHEMA.items())[:3]}}, ensure_ascii=False)\n",
        "print(\"스키마 JSON(앞 3개 필드):\", js[:150], \"...\")\n",
        "print(\"[입출력] 입력: 주간 직무별 인력 집계 15필드 / 출력: 결원 예측, 피로도·위험 지수, 보충 필요량 추천, 검토·결재 이력\")\n",
        "print(\"※ 가상 집계 자료이며 개인 단위 정보는 다루지 않는다. 실제 인사 자료의 외부 AI 서비스 입력은 금지된다.\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 4))\n",
        "v = df.assign(vr=df[\"vacancy\"] / df[\"authorized\"] * 100).groupby(\"week\")[\"vr\"].mean(); ax[0].plot(v.index, v.values); ax[0].set_title(\"Mean vacancy rate by week (%, synthetic)\"); ax[0].set_xlabel(\"week\")\n",
        "ax[1].bar([str(i) for i in range(len(s))], s[\"고피로비율\"] * 100, color=\"tab:orange\"); ax[1].set_title(\"High-fatigue rate by job (index 0-5, %)\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[스키마] 필드 15개, 필수 속성(type·unit·desc·source) 모두 정의된 필드 15개\n",
        "개인 식별 필드 0개(집계 인원 수만 사용)\n",
        "[정상 자료 검증] 표본 48건 오류 0건, 전체 2496건 오류 0건 (목표 오류 0건: 달성)\n",
        "\n",
        "[오류 주입 시험]\n",
        "  자료형 오류                -> 검출: authorized: 자료형 오류(str)\n",
        "  허용값 오류                -> 검출: job: 허용되지 않은 값 '미상직무'\n",
        "  범위 오류(음수 초과근무)        -> 검출: overtime_h: 최소값 0 미만\n",
        "  논리 오류(결원 불일치)         -> 검출: vacancy: 편제 - 현재원과 불일치\n",
        "  논리 오류(휴가+환자+교육>현재원)   -> 검출: leave+sick+training: 현재원 초과\n",
        "  필드 누락                 -> 검출: night_ratio: 필수 값 누락\n",
        "  결측값                   -> 검출: present: 필수 값 누락\n",
        "  비율 범위 초과              -> 검출: night_ratio: 최대값 1 초과\n",
        "주입 오류 8종 중 8종 검출\n",
        "\n",
        "[자료 구조 요약] 셀(부대 x 직무) 24개 x 104주 = 2496행\n",
        "      평균편제   결원율  평균초과근무  고피로비율\n",
        "job                            \n",
        "보급   28.00  0.18   18.81   0.52\n",
        "의무   29.25  0.07    8.31   0.25\n",
        "작전   35.75  0.06   12.69   0.34\n",
        "정비   31.50  0.09    9.95   0.35\n",
        "통신   25.75  0.14   16.21   0.50\n",
        "행정   32.75  0.06    9.82   0.29\n",
        "스키마 JSON(앞 3개 필드): {\"title\": \"adm01_input_schema\", \"fields\": {\"unit\": {\"type\": \"string\", \"enum\": [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"], \"unit\": \"-\", \"source\": \"인사 편제 집계\"} ...\n",
        "[입출력] 입력: 주간 직무별 인력 집계 15필드 / 출력: 결원 예측, 피로도·위험 지수, 보충 필요량 추천, 검토·결재 이력\n",
        "※ 가상 집계 자료이며 개인 단위 정보는 다루지 않는다. 실제 인사 자료의 외부 AI 서비스 입력은 금지된다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 02. 비정형 근무일지 파싱, 결원 사유 분류 및 근무 시간 정규화 전처리 모듈 구현\n",
        "**소요시간**: 약 2.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "근무일지 파싱, 결원 사유 분류 및 근무 시간 정규화 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **서식이 제각각인 비정형 근무일지를 파싱해 결원 사유를 분류하고 근무 시간을 시간 단위로 정규화하는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **시간 표기 5종('12시간 30분', '8h', '22:00~06:00', '오후 2시~익일 오전 6시', '총 9시간')이 섞인 가상 근무일지 6,000줄에서 부대·직무·근무 시간·상황·결원 사유(휴가·환자·교육·파견·전출·기타)를 정규식과 키워드로 뽑는다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/preprocess/adm01_preprocessor.py 및 전처리 전후 데이터셋` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 전술 환경 특유의 노이즈(스푸핑, 반사파, 센서 단절 등) 제거 타당성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 데이터 정제 처리 속도 1,000건/초 이상, 전처리 오류율 0%\n",
        "\n",
        "\n",
        "- 진행 기준: 정제 파이프라인 자동화 및 단위 테스트 케이스 5종 통과\n",
        "\n",
        "\n",
        "- 검증 방법: pytest 기반 전처리 정합성 테스트 및 결측치 보정율 확인\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"서식이 제각각인 가상 근무일지 문장 6,000줄(시간 표기 5종·결원 사유 표현 다수)을 정규식으로 파싱해 부대·직무·근무 시간(시간 단위)·주야·결원 사유를 표준 필드로 만들고, 주간·야간·훈련주간별 임계값을 자료로 조정해 잡음(이상 값·중복·잘린 줄)을 걸러내며 처리 속도(건/초)와 오류율을 측정한다.\"\"\"\n",
        "import re\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "REASONS = {\"휴가\": [\"휴가\", \"연가\", \"외박\"], \"환자\": [\"환자\", \"진료\", \"입원\", \"병가\"], \"교육·파견\": [\"교육\", \"파견\", \"훈련 참가\", \"학교\"], \"전출\": [\"전출\", \"전역\", \"전보\"]}\n",
        "SAY = {\"휴가\": [\"휴가\", \"연가 사용\", \"외박\"], \"환자\": [\"환자 발생\", \"진료\", \"병가\", \"입원\"], \"교육·파견\": [\"교육 입교\", \"파견\", \"학교 교육\"], \"전출\": [\"전출\", \"전역 예정\", \"전보\"], \"기타\": [\"사유 불명\", \"확인 중\", \"-\"]}\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 근무일지 생성(실제 부대·인원 아님). 줄마다 정답(truth)을 함께 만들어 파싱 정확도를 점검한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "rng = np.random.default_rng(12)\n",
        "\n",
        "\n",
        "def make_logs(n=6000, noise=0.03):\n",
        "    lines, truth = [], []\n",
        "    for i in range(n):\n",
        "        u, j = UNITS[rng.integers(4)], JOBS[rng.integers(6)]\n",
        "        ctx = rng.choice([\"평시\", \"야간\", \"훈련주간\"], p=[.55, .3, .15])\n",
        "        base = {\"평시\": 9, \"야간\": 11, \"훈련주간\": 15}[ctx]\n",
        "        h = round(float(np.clip(rng.normal(base, 1.3), 4, 20)) * 2) / 2\n",
        "        reason = rng.choice(list(SAY), p=[.3, .2, .2, .15, .15])\n",
        "        say = rng.choice(SAY[reason])\n",
        "        fmt = i % 5\n",
        "        us = u if rng.random() < .6 else u.replace(\"가상\", \"\")\n",
        "        if fmt == 0:\n",
        "            t = f\"[근무일지] 2026-{rng.integers(1, 13):02d}-{rng.integers(1, 28):02d} {us} {j} 근무 {int(h)}시간{'30분' if h % 1 else ''}({ctx}) 결원사유: {say}\"\n",
        "        elif fmt == 1:\n",
        "            t = f\"{us}/{j}/W{rng.integers(1, 53)} 근무 {h:g}h {ctx} / 사유 {say}\"\n",
        "        elif fmt == 2:\n",
        "            st = int(rng.integers(0, 24)); en = (st + int(h)) % 24\n",
        "            t = f\"{us} {j} 근무 {st:02d}:00~{en:02d}:{'30' if h % 1 else '00'} [{ctx}] 결원: {say}\"\n",
        "            h = int(h) + (0.5 if h % 1 else 0)\n",
        "        elif fmt == 3:\n",
        "            st = int(rng.integers(1, 12)); ap = rng.choice([\"오전\", \"오후\"]); total = st + (12 if ap == \"오후\" else 0) + int(h); en = total % 24\n",
        "            ap2, en12 = (\"오후\" if en >= 12 else \"오전\"), (en % 12 or 12)\n",
        "            t = f\"{us} {j} {ap} {st}시~{'익일 ' if total >= 24 else ''}{ap2} {en12}시 근무({ctx}), 결원 원인 {say}\"\n",
        "            h = float(int(h))\n",
        "        else:\n",
        "            t = f\"{us} {j} {ctx} 근무 총 {h:g}시간 사유={say}\"\n",
        "        truth.append((u, j, h, ctx, reason))\n",
        "        lines.append(t)\n",
        "    k = int(n * noise)\n",
        "    for idx in rng.choice(n, k, replace=False):                    # 잡음 주입: 이상 값(99시간·0.5시간)·잘린 줄\n",
        "        has_h = bool(re.search(r\"\\d(?:시간|h\\b)\", lines[idx]))\n",
        "        if idx % 3 == 0 and has_h:\n",
        "            lines[idx] = re.sub(r\"\\d+(?:\\.\\d+)?(?=\\s*(?:시간|h\\b))\", \"99\", lines[idx], count=1); truth[idx] = None\n",
        "        elif idx % 3 == 1 and has_h:\n",
        "            lines[idx] = re.sub(r\"\\d+(?:\\.\\d+)?(?=\\s*(?:시간|h\\b))\", \"0.5\", lines[idx], count=1); truth[idx] = None\n",
        "        else:\n",
        "            lines[idx] = lines[idx][: len(lines[idx]) // 3]; truth[idx] = None\n",
        "    return lines, truth\n",
        "\n",
        "\n",
        "lines, truth = make_logs()\n",
        "lines += lines[:60]; truth += truth[:60]                           # 중복 줄 60개 추가\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 파서: 부대·직무·근무 시간·상황·사유를 줄에서 뽑는다\n",
        "# ---------------------------------------------------------------------------\n",
        "RE_UNIT = re.compile(r\"(?:가상)?([A-D])대대\")\n",
        "RE_JOB = re.compile(\"|\".join(JOBS))\n",
        "RE_HM = re.compile(r\"(\\d+(?:\\.\\d+)?)\\s*시간(?:\\s*(\\d+)\\s*분)?\")\n",
        "RE_H = re.compile(r\"(\\d+(?:\\.\\d+)?)\\s*h\\b\")\n",
        "RE_CLK = re.compile(r\"(\\d{1,2}):(\\d{2})\\s*~\\s*(\\d{1,2}):(\\d{2})\")\n",
        "RE_KO = re.compile(r\"(오전|오후)\\s*(\\d{1,2})시\\s*~\\s*(?:익일\\s*)?(오전|오후)\\s*(\\d{1,2})시\")\n",
        "RE_CTX = re.compile(r\"훈련주간|야간|평시\")\n",
        "\n",
        "\n",
        "def parse_hours(t):\n",
        "    m = RE_CLK.search(t)\n",
        "    if m:\n",
        "        a, b = int(m[1]) * 60 + int(m[2]), int(m[3]) * 60 + int(m[4])\n",
        "        return ((b - a) % 1440) / 60\n",
        "    m = RE_KO.search(t)\n",
        "    if m:\n",
        "        a = int(m[2]) % 12 + (12 if m[1] == \"오후\" else 0); e = int(m[4]) % 12 + (12 if m[3] == \"오후\" else 0)\n",
        "        return float((e - a) % 24)\n",
        "    m = RE_HM.search(t)\n",
        "    if m:\n",
        "        return float(m[1]) + (int(m[2]) / 60 if m[2] else 0)\n",
        "    m = RE_H.search(t)\n",
        "    if m:\n",
        "        return float(m[1])\n",
        "    raise ValueError(\"시간 표기 해석 불가\")\n",
        "\n",
        "\n",
        "def classify(t):\n",
        "    tail = re.split(r\"결원사유|결원 원인|결원:|사유\", t)[-1]\n",
        "    for r, kws in REASONS.items():\n",
        "        if any(k in tail for k in kws):\n",
        "            return r\n",
        "    return \"기타\"\n",
        "\n",
        "\n",
        "def parse(t):\n",
        "    u, j, c = RE_UNIT.search(t), RE_JOB.search(t), RE_CTX.search(t)\n",
        "    if not (u and j and c):\n",
        "        raise ValueError(\"필수 항목(부대·직무·상황) 추출 실패\")\n",
        "    return {\"unit\": f\"가상{u[1]}대대\", \"job\": j[0], \"hours\": parse_hours(t), \"ctx\": c[0], \"reason\": classify(t)}\n",
        "\n",
        "\n",
        "# 1) 파싱 + 처리 속도\n",
        "t0 = time.perf_counter()\n",
        "ok, bad = [], []\n",
        "for i, t in enumerate(lines):\n",
        "    try:\n",
        "        ok.append({**parse(t), \"i\": i})\n",
        "    except ValueError as e:\n",
        "        bad.append((i, str(e)))\n",
        "dt = time.perf_counter() - t0\n",
        "df = pd.DataFrame(ok)\n",
        "print(f\"[파싱] {len(lines)}줄 중 성공 {len(df)}줄, 해석 실패(잘린 줄) {len(bad)}줄 -> 격리 / 속도 {len(lines) / dt:,.0f}건/초 (목표 1,000건/초 이상: {'달성' if len(lines) / dt >= 1000 else '미달'})\")\n",
        "\n",
        "# 2) 중복 정리와 잡음 제거: 상황별(평시·야간·훈련주간) 임계값을 자료의 중앙값±MAD로 조정\n",
        "n0 = len(df)\n",
        "df = df[~pd.Series([lines[i] for i in df[\"i\"]]).duplicated().to_numpy()]\n",
        "print(f\"중복 줄 {n0 - len(df)}개 제거 -> {len(df)}줄\")\n",
        "FIXED = 14.0                                                           # 고정 임계값(모든 상황 공통): 14시간 초과는 이상\n",
        "fixed_flag = df[\"hours\"] > FIXED\n",
        "thr = {}\n",
        "for c, g in df[df[\"hours\"] < 30].groupby(\"ctx\"):\n",
        "    med = g[\"hours\"].median(); mad = (g[\"hours\"] - med).abs().median() * 1.4826\n",
        "    thr[c] = (med - 4 * mad, med + 4 * mad)\n",
        "dyn_flag = pd.Series([not (thr[c][0] <= h <= thr[c][1]) for c, h in zip(df[\"ctx\"], df[\"hours\"])], index=df.index) | (df[\"hours\"] >= 30)\n",
        "print(\"상황별 동적 임계값(시간):\", \", \".join(f\"{c} {lo:.1f}~{hi:.1f}\" for c, (lo, hi) in thr.items()))\n",
        "tr = [truth[i] for i in df[\"i\"]]\n",
        "is_noise = np.array([t is None for t in tr])\n",
        "for name, fl in [(\"고정 임계값(14시간)\", fixed_flag.to_numpy()), (\"상황별 동적 임계값\", dyn_flag.to_numpy())]:\n",
        "    tp = int((fl & is_noise).sum()); fp = int((fl & ~is_noise).sum())\n",
        "    print(f\"  {name:16s}: 이상 판정 {int(fl.sum())}건, 실제 잡음 중 {tp}/{int(is_noise.sum())} 검출, 정상인데 이상 판정(오탐) {fp}건\")\n",
        "\n",
        "# 3) 파싱 정확도(정답과 비교): 잡음이 아닌 줄에서 시간·사유·부대·직무 일치율\n",
        "clean = df[~dyn_flag].copy()\n",
        "tc = [truth[i] for i in clean[\"i\"]]\n",
        "errs = sum((c[\"unit\"], c[\"job\"], c[\"ctx\"], c[\"reason\"]) != (t[0], t[1], t[3], t[4]) or abs(c[\"hours\"] - t[2]) > 0.01 for (_, c), t in zip(clean.iterrows(), tc) if t is not None)\n",
        "n_eval = sum(t is not None for t in tc)\n",
        "print(f\"\\n[전처리 정확도] 정제 후 {len(clean)}줄(정답 대조 {n_eval}줄) 중 불일치 {errs}줄 -> 전처리 오류율 {errs / n_eval:.2%} (목표 0%: {'달성' if errs == 0 else '미달'})\")\n",
        "print(\"[결원 사유 분류]\", clean[\"reason\"].value_counts().to_dict())\n",
        "print(\"[근무 시간 정규화] 상황별 평균 근무 시간:\", clean.groupby(\"ctx\")[\"hours\"].mean().round(1).to_dict())\n",
        "agg = clean.groupby([\"unit\", \"job\"]).agg(줄수=(\"hours\", \"size\"), 평균근무시간=(\"hours\", \"mean\")).round(1)\n",
        "print(\"[전처리 후 부대·직무 집계 예시]\"); print(agg.head(4).to_string())\n",
        "print(\"사람 확인 대상: 해석 실패 %d줄(예: %r), 사유 '기타' %d줄\" % (len(bad), lines[bad[0][0]][:24] if bad else \"\", int((clean[\"reason\"] == \"기타\").sum())))\n",
        "print(\"※ 임계값은 상황별 중앙값±4MAD로 자료에서 조정하며, 이상 판정된 줄은 삭제하지 않고 격리해 담당자가 확인한다.\")\n",
        "\n",
        "plt.figure(figsize=(8, 4))\n",
        "for c, g in clean.groupby(\"ctx\"):\n",
        "    plt.hist(g[\"hours\"], bins=np.arange(2, 22, 1), alpha=.5, label={\"평시\": \"normal\", \"야간\": \"night\", \"훈련주간\": \"training\"}[c])\n",
        "plt.axvline(FIXED, color=\"k\", ls=\"--\"); plt.legend(); plt.xlabel(\"duty hours\"); plt.title(\"Normalized duty hours by context (dashed: fixed threshold)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[파싱] 6060줄 중 성공 5953줄, 해석 실패(잘린 줄) 107줄 -> 격리 / 속도 56,271건/초 (목표 1,000건/초 이상: 달성)\n",
        "중복 줄 118개 제거 -> 5835줄\n",
        "상황별 동적 임계값(시간): 야간 5.1~16.9, 평시 3.1~14.9, 훈련주간 9.1~20.9\n",
        "  고정 임계값(14시간)    : 이상 판정 623건, 실제 잡음 중 38/75 검출, 정상인데 이상 판정(오탐) 585건\n",
        "  상황별 동적 임계값      : 이상 판정 75건, 실제 잡음 중 75/75 검출, 정상인데 이상 판정(오탐) 0건\n",
        "\n",
        "[전처리 정확도] 정제 후 5760줄(정답 대조 5760줄) 중 불일치 0줄 -> 전처리 오류율 0.00% (목표 0%: 달성)\n",
        "[결원 사유 분류] {'휴가': 1719, '환자': 1175, '교육·파견': 1161, '전출': 889, '기타': 816}\n",
        "[근무 시간 정규화] 상황별 평균 근무 시간: {'야간': 11.0, '평시': 8.9, '훈련주간': 14.9}\n",
        "[전처리 후 부대·직무 집계 예시]\n",
        "            줄수  평균근무시간\n",
        "unit  job             \n",
        "가상A대대 보급   226    10.4\n",
        "      의무   245    10.5\n",
        "      작전   235    10.3\n",
        "      정비   239    10.1\n",
        "사람 확인 대상: 해석 실패 107줄(예: '가상D대대 통신 근무 2'), 사유 '기타' 816줄\n",
        "※ 임계값은 상황별 중앙값±4MAD로 자료에서 조정하며, 이상 판정된 줄은 삭제하지 않고 격리해 담당자가 확인한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 03. 직무별 결원율 및 초과근무 시간 기반 1차 피로도 산출 Baseline 모델 구현\n",
        "**소요시간**: 약 2.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "직무별 결원율 및 초과근무 기반 피로도 산출 Baseline 모델을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **직무별 결원율과 초과근무 시간(야간 근무 비율 포함)으로 1차 피로도 점수를 계산하는 Baseline 모델을 만들고 가중치를 1차 튜닝해 정확도를 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **결원율·초과근무·야간 비율을 한도로 나누어 가중합한 피로도 점수(0~100)를 정의하고, 앞 52주로 가중치와 기준 점수를 격자 탐색한 뒤 뒤 52주로 정확도·정밀도·재현율·F1을 시험 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/models/adm01_baseline_model.py 및 기본 추론 결과 보고서` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 알고리즘의 논리적 흐름 및 군사적 판단 기준의 투명성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 기본 데이터셋 대상 탐지/처리 정확도 75% 이상 달성\n",
        "\n",
        "\n",
        "- 진행 기준: 단위 기능 동작 확인 및 기본 시나리오 실행 로그 생성\n",
        "\n",
        "\n",
        "- 검증 방법: 기준 벤치마크 데이터셋 대상 추론 결과 오차율 비교\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"직무별 결원율·초과근무 시간·야간 근무 비율로 1차 피로도 점수(0~100)를 계산하는 가중합 Baseline을 만들고, 초기 가중치와 격자 탐색으로 1차 튜닝한 가중치의 정확도·정밀도·재현율·판단 지연시간을 시험 기간 자료로 검증하며 부대·직무 셀별 피로도 순위와 판단 근거를 출력한다.\"\"\"\n",
        "import time\n",
        "import itertools\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "df = make_manning(7, 104)\n",
        "df[\"vac_rate\"] = df[\"vacancy\"] / df[\"authorized\"]\n",
        "tr, te = df[df[\"week\"] <= 52], df[df[\"week\"] > 52]                       # 앞 52주 학습·튜닝 / 뒤 52주 시험\n",
        "X = lambda d: np.column_stack([np.minimum(d[\"vac_rate\"] / 0.25, 1.2), np.minimum(d[\"overtime_h\"] / 20, 1.2), np.minimum(d[\"night_ratio\"] / 0.4, 1.2)])\n",
        "\n",
        "\n",
        "def score(d, w):\n",
        "    \"\"\"피로도 점수 = 100 x (w1 x 결원 부담 + w2 x 초과근무 부담 + w3 x 야간 부담).\"\"\"\n",
        "    return np.minimum(100 * X(d) @ np.asarray(w), 100)\n",
        "\n",
        "\n",
        "def metrics(y, p):\n",
        "    tp, fp, fn = int(((p == 1) & (y == 1)).sum()), int(((p == 1) & (y == 0)).sum()), int(((p == 0) & (y == 1)).sum())\n",
        "    pr, rc = tp / max(tp + fp, 1), tp / max(tp + fn, 1)\n",
        "    return {\"acc\": float((p == y).mean()), \"prec\": pr, \"rec\": rc, \"f1\": 2 * pr * rc / max(pr + rc, 1e-9)}\n",
        "\n",
        "\n",
        "W0, CUT0 = (1 / 3, 1 / 3, 1 / 3), 50                                    # 초기(동일 가중) 휴리스틱\n",
        "rule0 = ((te[\"vac_rate\"] >= 0.20) | (te[\"overtime_h\"] >= 15)).astype(int).to_numpy()   # 가장 단순한 규칙: 결원율 20% 이상 또는 초과근무 15시간 이상\n",
        "y_tr, y_te = tr[\"fatigue_label\"].to_numpy(), te[\"fatigue_label\"].to_numpy()\n",
        "m0 = metrics(y_te, (score(te, W0) >= CUT0).astype(int))\n",
        "trivial = float((y_te == 0).mean())\n",
        "print(f\"[Baseline] 학습 {len(tr)}행 / 시험 {len(te)}행, 고피로 비율 {y_te.mean():.1%} (모두 '정상'이라 답해도 정확도 {trivial:.1%})\")\n",
        "mr = metrics(y_te, rule0)\n",
        "print(f\"단순 규칙(결원율 20% 이상 또는 초과근무 15시간 이상): 정확도 {mr['acc']:.1%}, 정밀도 {mr['prec']:.1%}, 재현율 {mr['rec']:.1%}, F1 {mr['f1']:.2f}\")\n",
        "print(f\"초기 가중(동일 1/3, 기준 {CUT0}점): 정확도 {m0['acc']:.1%}, 정밀도 {m0['prec']:.1%}, 재현율 {m0['rec']:.1%}, F1 {m0['f1']:.2f}\")\n",
        "\n",
        "# 1차 튜닝: 가중치(합 1)와 기준 점수를 학습 기간에서 F1이 최대가 되게 격자 탐색\n",
        "best = None\n",
        "Xtr = X(tr)\n",
        "for w1, w2 in itertools.product(np.arange(0.0, 1.01, 0.1), repeat=2):\n",
        "    if w1 + w2 > 1.0001:\n",
        "        continue\n",
        "    w = np.array([w1, w2, 1 - w1 - w2]); s = 100 * Xtr @ w\n",
        "    for cut in range(35, 71, 5):\n",
        "        f = metrics(y_tr, (s >= cut).astype(int))[\"f1\"]\n",
        "        if best is None or f > best[0]:\n",
        "            best = (f, tuple(float(x) for x in np.round(w, 1)), cut)\n",
        "W1, CUT1 = best[1], best[2]\n",
        "m1 = metrics(y_te, (score(te, W1) >= CUT1).astype(int))\n",
        "print(f\"튜닝 가중 {W1}(결원·초과근무·야간), 기준 {CUT1}점 (학습 F1 {best[0]:.2f})\")\n",
        "print(f\"시험 기간: 정확도 {m1['acc']:.1%}, 정밀도 {m1['prec']:.1%}, 재현율 {m1['rec']:.1%}, F1 {m1['f1']:.2f} (목표 정확도 75% 이상: {'달성' if m1['acc'] >= 0.75 else '미달'}, 단 단순 기준선 {trivial:.1%}보다 {'높음' if m1['acc'] > trivial else '낮음'})\")\n",
        "\n",
        "# 판단 지연시간: 1건씩 계산(실시간 조회 상황)\n",
        "rec = te.to_dict(\"records\")\n",
        "t0 = time.perf_counter()\n",
        "for r in rec[:1000]:\n",
        "    s = min(100 * (W1[0] * min(r[\"vac_rate\"] / .25, 1.2) + W1[1] * min(r[\"overtime_h\"] / 20, 1.2) + W1[2] * min(r[\"night_ratio\"] / .4, 1.2)), 100); _ = s >= CUT1\n",
        "lat = (time.perf_counter() - t0) / 1000 * 1000\n",
        "print(f\"판단 지연시간 평균 {lat:.4f}ms/건\")\n",
        "\n",
        "# 2) 직무별·셀별 피로도 순위 (최근 12주 평균)\n",
        "last = df[df[\"week\"] > 92].copy(); last[\"fatigue\"] = score(last, W1)\n",
        "by_job = last.groupby(\"job\")[\"fatigue\"].mean().sort_values(ascending=False)\n",
        "print(\"\\n[직무별 평균 피로도(최근 12주)]\", \", \".join(f\"{k} {v:.0f}\" for k, v in by_job.items()))\n",
        "cell = last.groupby([\"unit\", \"job\"]).agg(피로도=(\"fatigue\", \"mean\"), 결원율=(\"vac_rate\", \"mean\"), 초과근무=(\"overtime_h\", \"mean\"), 야간=(\"night_ratio\", \"mean\")).sort_values(\"피로도\", ascending=False)\n",
        "print(\"[피로도 상위 5개 셀과 판단 근거(기여 점수)]\")\n",
        "for (u, j), r in cell.head(5).iterrows():\n",
        "    c = 100 * np.array(W1) * np.minimum(np.array([r[\"결원율\"] / .25, r[\"초과근무\"] / 20, r[\"야간\"] / .4]), 1.2)\n",
        "    print(f\"  {u} {j}: 피로도 {r['피로도']:.0f} = 결원 {c[0]:.0f} + 초과근무 {c[1]:.0f} + 야간 {c[2]:.0f}  (결원율 {r['결원율']:.0%}, 초과근무 {r['초과근무']:.1f}h)\")\n",
        "n_hi = int((cell[\"피로도\"] >= CUT1).sum())\n",
        "print(f\"기준 {CUT1}점 이상 셀 {n_hi}/24개 -> 인사관이 우선 확인. 점수는 판단 보조이며 휴식·교대 조정은 지휘관·인사관이 결정한다.\")\n",
        "\n",
        "# 3) 오분류 분석 (튜닝 가중치 시험 기간)\n",
        "pred = (score(te, W1) >= CUT1).astype(int)\n",
        "fn = te[(pred == 0) & (y_te == 1)]\n",
        "print(f\"\\n[미탐 {len(fn)}건 특징] 평균 결원율 {fn['vac_rate'].mean():.0%}, 초과근무 {fn['overtime_h'].mean():.1f}h, 야간 {fn['night_ratio'].mean():.0%} -> 점수 경계 부근 사례(라벨에는 휴가·환자·교육 인원 비율과 잡음도 반영되어 이 3개 변수만으로는 설명되지 않음)\")\n",
        "\n",
        "plt.figure(figsize=(7, 4))\n",
        "plt.scatter(te[\"vac_rate\"] * 100, te[\"overtime_h\"], c=te[\"fatigue_label\"], cmap=\"coolwarm\", s=8, alpha=.6)\n",
        "plt.xlabel(\"vacancy rate (%)\"); plt.ylabel(\"overtime hours/person/week\"); plt.title(\"High-fatigue label (red) by vacancy and overtime\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[Baseline] 학습 1248행 / 시험 1248행, 고피로 비율 34.8% (모두 '정상'이라 답해도 정확도 65.2%)\n",
        "단순 규칙(결원율 20% 이상 또는 초과근무 15시간 이상): 정확도 92.7%, 정밀도 96.7%, 재현율 81.8%, F1 0.89\n",
        "초기 가중(동일 1/3, 기준 50점): 정확도 89.7%, 정밀도 78.9%, 재현율 95.9%, F1 0.87\n",
        "튜닝 가중 (0.3, 0.6, 0.1)(결원·초과근무·야간), 기준 55점 (학습 F1 0.91)\n",
        "시험 기간: 정확도 94.1%, 정밀도 91.9%, 재현율 91.0%, F1 0.91 (목표 정확도 75% 이상: 달성, 단 단순 기준선 65.2%보다 높음)\n",
        "판단 지연시간 평균 0.0051ms/건\n",
        "\n",
        "[직무별 평균 피로도(최근 12주)] 보급 68, 통신 64, 정비 52, 작전 51, 의무 44, 행정 42\n",
        "[피로도 상위 5개 셀과 판단 근거(기여 점수)]\n",
        "  가상B대대 보급: 피로도 100 = 결원 36 + 초과근무 72 + 야간 6  (결원율 57%, 초과근무 40.0h)\n",
        "  가상C대대 통신: 피로도 100 = 결원 36 + 초과근무 72 + 야간 7  (결원율 49%, 초과근무 39.9h)\n",
        "  가상A대대 정비: 피로도 96 = 결원 34 + 초과근무 71 + 야간 6  (결원율 28%, 초과근무 23.6h)\n",
        "  가상A대대 보급: 피로도 83 = 결원 20 + 초과근무 61 + 야간 6  (결원율 17%, 초과근무 20.3h)\n",
        "  가상D대대 통신: 피로도 82 = 결원 19 + 초과근무 68 + 야간 6  (결원율 16%, 초과근무 22.5h)\n",
        "기준 55점 이상 셀 10/24개 -> 인사관이 우선 확인. 점수는 판단 보조이며 휴식·교대 조정은 지휘관·인사관이 결정한다.\n",
        "\n",
        "[미탐 39건 특징] 평균 결원율 7%, 초과근무 11.7h, 야간 26% -> 점수 경계 부근 사례(라벨에는 휴가·환자·교육 인원 비율과 잡음도 반영되어 이 3개 변수만으로는 설명되지 않음)\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 04. 시계열 분석 기반 부서별 향후 3개월 결원 추세 및 보충 필요 인력 예측 모델 고도화\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "시계열 분석 기반 향후 3개월 결원 추세 및 보충 필요량 예측 모델을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **시계열 분석으로 부대·직무별 향후 3개월(13주) 결원 추세를 예측하고 보충 필요 인원을 산출하는 모델을 고도화해 보충 필요 셀 탐지 재현율을 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **24개 셀의 주간 결원율을 단순 예측, 릿지 회귀(특징 재설계 전후), 교대 보충·전출 집중기를 반영한 재고-흐름 모의(300경로)로 예측해 4개 시점 롤링 검증으로 오차를 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/models/adm01_advanced_detector.py 및 예외 탐지 테스트 결과` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 기만/교란 공격에 대한 방어 로직의 현실성 및 신뢰성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 복합 위협 탐지 재현율(Recall) 85% 이상, 처리 지연 200ms 이하\n",
        "\n",
        "\n",
        "- 진행 기준: 복합 위협 시나리오 10종에 대한 탐지 테스트 통과\n",
        "\n",
        "\n",
        "- 검증 방법: 합성 위협 데이터셋 주입 시 이상치 탐지 정확도 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"부대·직무 24개 셀의 주간 결원율을 단순 예측·릿지 회귀(특징 재설계 v1→v2)·재고-흐름 모의(교대 보충·전출 집중기 반영, 300경로)로 향후 13주(3개월) 예측해 4개 시점 롤링 검증으로 비교하고, 보충 필요 셀 탐지의 재현율·정밀도 균형(확률 기준 선택)과 처리 지연을 측정해 보충 필요 인원을 산출한다.\"\"\"\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "df = make_manning(7, 104)\n",
        "cells = cell_arrays(df)\n",
        "ORIGINS = [52, 65, 78, 91]                                  # 예측 시점(그 주까지의 자료만 사용)\n",
        "res = []\n",
        "lat, lat3 = [], []\n",
        "for o in ORIGINS:\n",
        "    m1, m2 = fit_fc(cells, o, v2=False), fit_fc(cells, o, v2=True)\n",
        "    for k, c in cells.items():\n",
        "        act = c[\"vr\"][o:o + H]\n",
        "        t0 = time.perf_counter(); p2 = forecast(m2, c, o, True); lat.append((time.perf_counter() - t0) * 1000)\n",
        "        p1 = forecast(m1, c, o, False)\n",
        "        naive = np.full(H, c[\"vr\"][o - 1])\n",
        "        t0 = time.perf_counter(); paths, _ = mc_forecast(c, o); lat3.append((time.perf_counter() - t0) * 1000)\n",
        "        res.append({\"origin\": o, \"cell\": k, \"auth\": c[\"auth\"], \"act_peak\": act.max(), \"act_end\": act[-1], \"naive\": naive, \"v1\": p1, \"v2\": p2, \"v3\": paths.mean(0), \"p_need\": float((paths.max(axis=1) >= NEED_CUT).mean()), \"act\": act})\n",
        "R = pd.DataFrame(res)\n",
        "mae = lambda col: float(np.mean([np.abs(r[col] - r[\"act\"]).mean() * r[\"auth\"] for _, r in R.iterrows()]))\n",
        "print(f\"[롤링 검증] 시점 {ORIGINS} x 셀 {len(cells)}개 = {len(R)}건, 예측 13주\")\n",
        "print(f\"결원 인원 MAE: 단순 예측(마지막 값 유지) {mae('naive'):.2f}명 / 회귀 v1 {mae('v1'):.2f}명 / 회귀 v2(특징 재설계) {mae('v2'):.2f}명 / 재고-흐름 모의 평균 {mae('v3'):.2f}명\")\n",
        "print(f\"예측 지연: 회귀 셀 1개 평균 {np.mean(lat):.1f}ms, 모의 300경로 평균 {np.mean(lat3):.1f}ms/99백분위 {np.percentile(lat3, 99):.1f}ms (24개 셀 전체 {np.mean(lat3) * 24:.0f}ms, 목표 200ms 이하: {'달성' if np.percentile(lat3, 99) <= 200 else '미달'})\")\n",
        "\n",
        "# 보충 필요 셀 탐지: 향후 13주 중 결원율이 NEED_CUT 이상이 되는가\n",
        "act_flag = R[\"act_peak\"] >= NEED_CUT\n",
        "def det(pred):\n",
        "    pred = np.asarray(pred); tp, fp, fn = int((pred & act_flag).sum()), int((pred & ~act_flag).sum()), int((~pred & act_flag).sum())\n",
        "    return tp / max(tp + fn, 1), tp / max(tp + fp, 1), fp\n",
        "print(f\"\\n[보충 필요(결원율 {NEED_CUT:.0%} 이상) 탐지] 실제 해당 {int(act_flag.sum())}건 / {len(R)}건\")\n",
        "rows = [(\"단순 예측\", [r[\"naive\"].max() >= NEED_CUT for _, r in R.iterrows()]), (\"회귀 v1\", [r[\"v1\"].max() >= NEED_CUT for _, r in R.iterrows()]), (\"회귀 v2\", [r[\"v2\"].max() >= NEED_CUT for _, r in R.iterrows()])]\n",
        "rows += [(f\"모의 확률 {p:.2f} 이상\", (R[\"p_need\"] >= p).to_list()) for p in (0.5, 0.3, 0.15)]\n",
        "for name, pred in rows:\n",
        "    rc, pr, fp = det(pred); print(f\"  {name:16s} 재현율 {rc:5.1%}, 정밀도 {pr:5.1%}, 오경보 {fp}건\")\n",
        "P0 = 0.15\n",
        "rc, pr, fp = det((R[\"p_need\"] >= P0).to_list())\n",
        "print(f\"-> 선택: 모의 확률 {P0} 이상을 '보충 필요 가능성'으로 표시, 재현율 {rc:.1%} (목표 85% 이상: {'달성' if rc >= 0.85 else '미달'}), 대신 정밀도 {pr:.0%}로 오경보 {fp}건을 사람이 걸러야 함\")\n",
        "\n",
        "# 오경보 패턴: 선택 기준의 오경보가 어디에 몰리는가\n",
        "fa = R[(R[\"p_need\"] >= P0) & ~act_flag]\n",
        "near = float((fa[\"act_peak\"] >= 0.10).mean()) if len(fa) else 0.0\n",
        "print(f\"[오경보 패턴] 오경보 {len(fa)}건 중 실제 최대 결원율이 10~15%인 경계 근처 사례 {near:.0%} -> 교대 보충 도착 시점·전출 규모의 불확실성이 원인, 확률이 낮은 건(0.15~0.5)은 '관찰'로 표시하고 0.5 이상만 '보충 필요'로 올린다\")\n",
        "hi = int(((R[\"p_need\"] >= 0.5) & act_flag).sum()); hin = int((R[\"p_need\"] >= 0.5).sum())\n",
        "print(f\"확률 0.5 이상 {hin}건 중 실제 보충 필요 {hi}건({hi / max(hin, 1):.0%})\")\n",
        "\n",
        "# 보충 필요 인원(경로의 90백분위 최대 결원 - 허용 결원 10%)과 실제 비교 (마지막 시점)\n",
        "last = R[R[\"origin\"] == ORIGINS[-1]].copy()\n",
        "need = lambda peak, a: max(0, int(np.ceil((peak - 0.10) * a)))\n",
        "q = []\n",
        "for _, r in last.iterrows():\n",
        "    paths, _ = mc_forecast(cells[r[\"cell\"]], ORIGINS[-1]); q.append(need(np.percentile(paths.max(axis=1), 75), r[\"auth\"]))\n",
        "last[\"pred_need\"] = q\n",
        "last[\"act_need\"] = [need(r[\"act\"].max(), r[\"auth\"]) for _, r in last.iterrows()]\n",
        "print(f\"\\n[시점 {ORIGINS[-1]}주 기준 보충 필요 인원(경로 최대 결원의 75백분위 기준)] 예측 합계 {int(last['pred_need'].sum())}명, 실제 합계 {int(last['act_need'].sum())}명, 셀별 평균 절대오차 {np.abs(last['pred_need'] - last['act_need']).mean():.2f}명\")\n",
        "top = last.sort_values(\"pred_need\", ascending=False).head(4)\n",
        "for _, r in top.iterrows():\n",
        "    print(f\"  {r['cell'][0]} {r['cell'][1]}: 예측 {r['pred_need']}명(실제 {r['act_need']}명), 보충 필요 확률 {r['p_need']:.0%}, 13주 말 예상 결원율 {r['v3'][-1]:.0%}\")\n",
        "print(\"※ 예측은 가상 자료 기반 판단 보조이며 보충 요청과 배치는 인사관이 결정한다. 모델이 모르는 대량 전출·파견은 AAR에서 재보정한다.\")\n",
        "\n",
        "best = top.iloc[0]\n",
        "plt.figure(figsize=(8, 4)); x = np.arange(H)\n",
        "plt.plot(x, best[\"act\"] * 100, \"k-\", label=\"actual\"); plt.plot(x, best[\"v3\"] * 100, \"b--\", label=\"stock-flow mean\"); plt.plot(x, best[\"v2\"] * 100, \"g-.\", label=\"ridge v2\"); plt.plot(x, best[\"naive\"] * 100, \"g:\", label=\"naive\")\n",
        "plt.axhline(NEED_CUT * 100, color=\"r\", ls=\":\"); plt.xlabel(\"weeks ahead\"); plt.ylabel(\"vacancy rate (%)\"); plt.title(\"13-week vacancy forecast (highest-need cell)\"); plt.legend(); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[롤링 검증] 시점 [52, 65, 78, 91] x 셀 24개 = 96건, 예측 13주\n",
        "결원 인원 MAE: 단순 예측(마지막 값 유지) 1.39명 / 회귀 v1 1.45명 / 회귀 v2(특징 재설계) 1.16명 / 재고-흐름 모의 평균 1.07명\n",
        "예측 지연: 회귀 셀 1개 평균 5.7ms, 모의 300경로 평균 1.9ms/99백분위 5.4ms (24개 셀 전체 45ms, 목표 200ms 이하: 달성)\n",
        "\n",
        "[보충 필요(결원율 15% 이상) 탐지] 실제 해당 35건 / 96건\n",
        "  단순 예측            재현율 45.7%, 정밀도 100.0%, 오경보 0건\n",
        "  회귀 v1            재현율 45.7%, 정밀도 100.0%, 오경보 0건\n",
        "  회귀 v2            재현율 54.3%, 정밀도 100.0%, 오경보 0건\n",
        "  모의 확률 0.50 이상    재현율 60.0%, 정밀도 87.5%, 오경보 3건\n",
        "  모의 확률 0.30 이상    재현율 71.4%, 정밀도 67.6%, 오경보 12건\n",
        "  모의 확률 0.15 이상    재현율 85.7%, 정밀도 55.6%, 오경보 24건\n",
        "-> 선택: 모의 확률 0.15 이상을 '보충 필요 가능성'으로 표시, 재현율 85.7% (목표 85% 이상: 달성), 대신 정밀도 56%로 오경보 24건을 사람이 걸러야 함\n",
        "[오경보 패턴] 오경보 24건 중 실제 최대 결원율이 10~15%인 경계 근처 사례 67% -> 교대 보충 도착 시점·전출 규모의 불확실성이 원인, 확률이 낮은 건(0.15~0.5)은 '관찰'로 표시하고 0.5 이상만 '보충 필요'로 올린다\n",
        "확률 0.5 이상 24건 중 실제 보충 필요 21건(88%)\n",
        "\n",
        "[시점 91주 기준 보충 필요 인원(경로 최대 결원의 75백분위 기준)] 예측 합계 77명, 실제 합계 71명, 셀별 평균 절대오차 1.00명\n",
        "  가상C대대 통신: 예측 17명(실제 15명), 보충 필요 확률 100%, 13주 말 예상 결원율 49%\n",
        "  가상B대대 보급: 예측 16명(실제 16명), 보충 필요 확률 100%, 13주 말 예상 결원율 51%\n",
        "  가상A대대 정비: 예측 7명(실제 8명), 보충 필요 확률 100%, 13주 말 예상 결원율 22%\n",
        "  가상C대대 정비: 예측 4명(실제 3명), 보충 필요 확률 58%, 13주 말 예상 결원율 14%\n",
        "※ 예측은 가상 자료 기반 판단 보조이며 보충 요청과 배치는 인사관이 결정한다. 모델이 모르는 대량 전출·파견은 AAR에서 재보정한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 05. 인력 부족 위험 지수 산출 및 인사 배치 추천 신뢰도 엔진 개발\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "인력 부족 위험 지수 산출 및 인사 배치 추천 신뢰도 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **인력 부족 위험 지수를 산출하고 신뢰도(High/Med/Low) 등급과 함께 인사 배치(인원 이동) 추천을 만드는 신뢰도 엔진을 개발한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **보충 필요 확률·예상 최대 결원율·초과근무 부담으로 위험 지수를, 예측 경로 분산·모델 간 일치·자료 충분도로 신뢰도를 계산해 가중치 매트릭스로 보여 준다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/verification/adm01_confidence_engine.py` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 신뢰도 등급(High/Med/Low) 부여 기준의 군사적 타당성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 오경보율(FAR) 40% 이상 감축, 신뢰도 스코어 산출 일관성 90% 이상\n",
        "\n",
        "\n",
        "- 진행 기준: 오경보 필터링 전후 비교 지표 산출 및 통합 테스트 완료\n",
        "\n",
        "\n",
        "- 검증 방법: ROC-AUC 곡선 분석 및 임계값 민감도 테스트\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"부대·직무 셀별 인력 부족 위험 지수(0~100)와 신뢰도(High/Med/Low: 예측 분산·모델 간 일치·자료 충분도 가중합)를 산출하고, 신뢰도 낮은 경보를 '확인 요청'으로 분리해 단순 경보 대비 오경보율 감축과 같은 설정 반복 실행 시 등급 일관성을 검증하며 신뢰도를 반영한 인사 배치(인원 이동) 추천안을 만든다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "df = make_manning(7, 104)\n",
        "cells = cell_arrays(df)\n",
        "ot = {k: g.sort_values(\"week\")[\"overtime_h\"].to_numpy() for k, g in df.groupby([\"unit\", \"job\"])}\n",
        "CONF_W = {\"spread\": 0.5, \"agree\": 0.3, \"data\": 0.2}                     # 신뢰도 가중치 매트릭스(합 1)\n",
        "RISK_W = {\"prob\": 0.5, \"peak\": 0.3, \"load\": 0.2}                        # 위험 지수 가중치(합 1)\n",
        "GRADE = [(60, \"높음\"), (35, \"보통\"), (0, \"낮음\")]\n",
        "\n",
        "\n",
        "def assess(k, origin, seed=0, models=None):\n",
        "    c = cells[k]\n",
        "    paths, par = mc_forecast(c, origin, n_path=800, seed=seed)\n",
        "    peaks = paths.max(axis=1)\n",
        "    p_need = float((peaks >= NEED_CUT).mean())\n",
        "    load = float(np.clip(ot[k][origin - 4:origin].mean() / 20, 0, 1))                                   # 최근 4주 초과근무 부담\n",
        "    risk = 100 * (RISK_W[\"prob\"] * p_need + RISK_W[\"peak\"] * min(peaks.mean() / 0.30, 1) + RISK_W[\"load\"] * load)\n",
        "    iqr = float(np.percentile(peaks, 75) - np.percentile(peaks, 25))\n",
        "    v2_peak = forecast(models[origin], c, origin, True).max()                                         # 서로 다른 모델(회귀)과의 일치\n",
        "    agree = 1 - min(abs(v2_peak - peaks.mean()) / 0.10, 1)\n",
        "    conf = CONF_W[\"spread\"] * (1 - min(iqr / 0.12, 1)) + CONF_W[\"agree\"] * agree + CONF_W[\"data\"] * min(origin / 52, 1)\n",
        "    return {\"cell\": k, \"origin\": origin, \"risk\": risk, \"p_need\": p_need, \"peak\": float(peaks.mean()), \"iqr\": iqr, \"agree\": agree, \"conf\": conf,\n",
        "            \"grade\": next(g for t, g in GRADE if risk >= t), \"level\": \"High\" if conf >= 0.70 else \"Med\" if conf >= 0.45 else \"Low\"}\n",
        "\n",
        "\n",
        "ORIGINS = [52, 65, 78, 91]\n",
        "models = {o: fit_fc(cells, o, True) for o in ORIGINS}\n",
        "rows = [assess(k, o, models=models) for o in ORIGINS for k in cells]\n",
        "A = pd.DataFrame(rows)\n",
        "A[\"act_need\"] = [cells[r.cell][\"vr\"][r.origin:r.origin + H].max() >= NEED_CUT for r in A.itertuples()]\n",
        "A[\"base_alarm\"] = [(cells[r.cell][\"vr\"][r.origin - 1] >= 0.10) or (ot[r.cell][r.origin - 4:r.origin].mean() >= 15) for r in A.itertuples()]\n",
        "A[\"alarm\"] = (A[\"risk\"] >= 50) & (A[\"level\"] != \"Low\")                               # 신뢰도 낮은 경보는 '확인 요청'으로 분리\n",
        "A[\"check\"] = (A[\"risk\"] >= 50) & (A[\"level\"] == \"Low\")\n",
        "far = lambda m: float(((m) & ~A[\"act_need\"]).sum() / max(m.sum(), 1))\n",
        "fa0, fa1 = far(A[\"base_alarm\"]), far(A[\"alarm\"])\n",
        "print(f\"[경보 비교] 셀 x 시점 {len(A)}건, 실제 보충 필요 {int(A['act_need'].sum())}건\")\n",
        "print(f\"  단순 경보(결원율 10% 이상 또는 초과근무 15h 이상): 경보 {int(A['base_alarm'].sum())}건, 오경보율 {fa0:.1%}, 재현율 {float((A['base_alarm'] & A['act_need']).sum() / A['act_need'].sum()):.1%}\")\n",
        "print(f\"  신뢰도 반영 경보(위험 50 이상 + 신뢰도 Med 이상): 경보 {int(A['alarm'].sum())}건, 오경보율 {fa1:.1%}, 재현율 {float((A['alarm'] & A['act_need']).sum() / A['act_need'].sum()):.1%}, 확인 요청 {int(A['check'].sum())}건\")\n",
        "print(f\"  오경보율 {fa0:.1%} -> {fa1:.1%}: {(fa0 - fa1) / max(fa0, 1e-9):.0%} 감축 (목표 40% 이상: {'달성' if (fa0 - fa1) / max(fa0, 1e-9) >= 0.4 else '미달'})\")\n",
        "\n",
        "# 점수 산출 일관성: 같은 입력으로 난수 시드만 바꿔 5회 산출했을 때 위험 등급이 같은 셀의 비율\n",
        "last = [assess(k, 91, seed=s, models=models) for s in range(5) for k in cells]\n",
        "G = pd.DataFrame(last)\n",
        "cons = G.groupby(\"cell\")[\"grade\"].nunique().eq(1).mean(); consl = G.groupby(\"cell\")[\"level\"].nunique().eq(1).mean()\n",
        "print(f\"\\n[일관성] 시드 5회 반복: 위험 등급 동일 셀 {cons:.0%}, 신뢰도 등급 동일 셀 {consl:.0%}, 위험 지수 표준편차 평균 {G.groupby('cell')['risk'].std().mean():.2f}점 (목표 90% 이상: {'달성' if cons >= 0.9 else '미달'})\")\n",
        "print(\"가중치 매트릭스: 위험 지수\", RISK_W, \"/ 신뢰도\", CONF_W, \"/ 등급 기준 위험\", GRADE, \"신뢰도 High 0.70, Med 0.45\")\n",
        "\n",
        "# 현재 시점(91주) 위험 순위와 불확실성 표시\n",
        "now = A[A[\"origin\"] == 91].sort_values(\"risk\", ascending=False)\n",
        "print(\"\\n[위험 지수 상위 6개 셀(91주 시점)]  막대: 최대 결원율 평균, 구간: 경로 IQR\")\n",
        "for r in now.head(6).itertuples():\n",
        "    bar = \"#\" * int(r.peak * 40)\n",
        "    print(f\"  {r.cell[0]} {r.cell[1]}: 위험 {r.risk:4.0f}({r.grade}) 신뢰도 {r.conf:.2f}({r.level}) 보충 필요 확률 {r.p_need:.0%} {bar} +-{r.iqr / 2:.0%}\")\n",
        "\n",
        "# 인사 배치 추천: 위험 높음 셀로 같은 직무의 여유 셀(위험 낮음, 결원 적음)에서 이동\n",
        "rec = []\n",
        "for r in now[now[\"grade\"] == \"높음\"].itertuples():\n",
        "    k = r.cell; need = max(1, int(np.ceil((r.peak - 0.10) * cells[k][\"auth\"])))\n",
        "    pool = [s for s in now[(now[\"grade\"] == \"낮음\")].itertuples() if s.cell[1] == k[1] and s.cell[0] != k[0]]\n",
        "    for s in sorted(pool, key=lambda s: s.peak)[:1]:\n",
        "        sp = int(max(0, np.floor((0.08 - s.peak) * cells[s.cell][\"auth\"])))\n",
        "        mv = min(need, max(sp, 1))\n",
        "        rec.append((k, s.cell, mv, need, \"High\" if min(r.conf, s.conf) >= 0.7 else \"Med\" if min(r.conf, s.conf) >= 0.45 else \"Low\"))\n",
        "print(\"\\n[배치 추천안: 같은 직무 여유 셀 -> 부족 셀, 신뢰도는 두 셀 중 낮은 값]\")\n",
        "for to, fr, mv, need, cl in rec:\n",
        "    print(f\"  {fr[0]}({fr[1]}) -> {to[0]}({to[1]}): {mv}명 이동 제안 (필요 {need}명), 신뢰도 {cl}\" + (\"  <- 사람 확인 필수\" if cl != \"High\" else \"\"))\n",
        "if not rec:\n",
        "    print(\"  해당 직무에 여유 셀이 없어 외부 보충 요청 대상\")\n",
        "print(\"※ 위험 지수·신뢰도는 판단 보조 지표이며 통계적 신뢰구간이 아니다. 인원 배치는 인사관이 검토·승인한다(M7).\")\n",
        "\n",
        "fig, ax = plt.subplots(figsize=(8, 4))\n",
        "t = now.head(8); ax.barh([f\"{r.cell[0][2]}{i + 1}\" for i, r in enumerate(t.itertuples())], t[\"risk\"], xerr=t[\"iqr\"] * 100 / 2, color=[\"tab:red\" if g == \"높음\" else \"tab:orange\" if g == \"보통\" else \"tab:green\" for g in t[\"grade\"]])\n",
        "ax.invert_yaxis(); ax.set_xlabel(\"risk index (0-100), error bar = forecast IQR\"); ax.set_title(\"Top-8 risk cells (label: unit letter + rank)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[경보 비교] 셀 x 시점 96건, 실제 보충 필요 35건\n",
        "  단순 경보(결원율 10% 이상 또는 초과근무 15h 이상): 경보 33건, 오경보율 36.4%, 재현율 60.0%\n",
        "  신뢰도 반영 경보(위험 50 이상 + 신뢰도 Med 이상): 경보 18건, 오경보율 11.1%, 재현율 45.7%, 확인 요청 6건\n",
        "  오경보율 36.4% -> 11.1%: 69% 감축 (목표 40% 이상: 달성)\n",
        "\n",
        "[일관성] 시드 5회 반복: 위험 등급 동일 셀 96%, 신뢰도 등급 동일 셀 88%, 위험 지수 표준편차 평균 0.74점 (목표 90% 이상: 달성)\n",
        "가중치 매트릭스: 위험 지수 {'prob': 0.5, 'peak': 0.3, 'load': 0.2} / 신뢰도 {'spread': 0.5, 'agree': 0.3, 'data': 0.2} / 등급 기준 위험 [(60, '높음'), (35, '보통'), (0, '낮음')] 신뢰도 High 0.70, Med 0.45\n",
        "\n",
        "[위험 지수 상위 6개 셀(91주 시점)]  막대: 최대 결원율 평균, 구간: 경로 IQR\n",
        "  가상B대대 보급: 위험  100(높음) 신뢰도 0.83(High) 보충 필요 확률 100% ####################### +-0%\n",
        "  가상C대대 통신: 위험  100(높음) 신뢰도 0.47(Med) 보충 필요 확률 100% ###################### +-3%\n",
        "  가상A대대 정비: 위험   94(높음) 신뢰도 0.64(Med) 보충 필요 확률 100% ########## +-3%\n",
        "  가상D대대 통신: 위험   67(높음) 신뢰도 0.45(Low) 보충 필요 확률 62% ####### +-5%\n",
        "  가상A대대 보급: 위험   63(높음) 신뢰도 0.52(Med) 보충 필요 확률 60% ####### +-5%\n",
        "  가상C대대 정비: 위험   54(보통) 신뢰도 0.51(Med) 보충 필요 확률 54% ####### +-4%\n",
        "\n",
        "[배치 추천안: 같은 직무 여유 셀 -> 부족 셀, 신뢰도는 두 셀 중 낮은 값]\n",
        "  가상D대대(보급) -> 가상B대대(보급): 1명 이동 제안 (필요 16명), 신뢰도 High\n",
        "  가상B대대(통신) -> 가상C대대(통신): 1명 이동 제안 (필요 16명), 신뢰도 Med  <- 사람 확인 필수\n",
        "  가상D대대(정비) -> 가상A대대(정비): 1명 이동 제안 (필요 6명), 신뢰도 Med  <- 사람 확인 필수\n",
        "  가상B대대(통신) -> 가상D대대(통신): 1명 이동 제안 (필요 3명), 신뢰도 Low  <- 사람 확인 필수\n",
        "  가상D대대(보급) -> 가상A대대(보급): 1명 이동 제안 (필요 3명), 신뢰도 Med  <- 사람 확인 필수\n",
        "※ 위험 지수·신뢰도는 판단 보조 지표이며 통계적 신뢰구간이 아니다. 인원 배치는 인사관이 검토·승인한다(M7).\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 06. 지휘관/인사참모용 부대 근무부하 현황 및 결원 예측 가시화 대시보드 구현\n",
        "**소요시간**: 약 2.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "부대 근무부하 현황 및 결원 예측 가시화 대시보드를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **지휘관·인사참모가 부대 근무부하 현황과 결원 예측을 한눈에 볼 수 있는 가시화 대시보드를 구현하고 응답 성능을 측정한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **부대 x 직무 24개 셀의 현재 결원율·13주 후 예측·초과근무·피로도·보충 필요 확률을 상태(긴급·주의·정상) 강조와 함께 표와 히트맵으로 만들고 위험 셀의 추세·예측 구간 그래프를 그린다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/ui/adm01_dashboard.py (또는 웹 UI 컴포넌트)` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 전술 작전 상황판으로서의 직관성 및 인지 부하 감소도\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 화면 갱신 주기 1초 이내, UI 응답성 100ms 이내\n",
        "\n",
        "\n",
        "- 진행 기준: 주요 지표 시각화 컴포넌트 4종 이상 렌더링 및 실시간 데이터 연동\n",
        "\n",
        "\n",
        "- 검증 방법: 사용자 인터랙션 테스트 및 모의 데이터 스트림 렌더링 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"부대·직무 셀의 현재 결원율·초과근무·피로도와 향후 13주 결원 예측·보충 필요 확률을 지휘관·인사참모용 현황판(표·히트맵·추세 그래프, 긴급 경보 강조, 야간 모드)으로 보여 주고, 전체 갱신 시간(1초 이내)과 필터·드릴다운·모드 전환 응답 시간(100ms 이내)을 측정한다.\"\"\"\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "W1, CUT1 = (0.3, 0.6, 0.1), 55                                         # 피로도 가중치·기준(M3 튜닝 결과 예시)\n",
        "PALETTE = {\"day\": {\"bg\": \"white\", \"fg\": \"black\", \"cmap\": \"YlOrRd\", \"alert\": \"red\"}, \"night\": {\"bg\": \"#10151c\", \"fg\": \"#d8dee9\", \"cmap\": \"magma\", \"alert\": \"#ffb000\"}}\n",
        "\n",
        "\n",
        "def refresh(df):\n",
        "    \"\"\"원 자료에서 현황판 데이터를 모두 다시 계산한다(전체 갱신).\"\"\"\n",
        "    cells = cell_arrays(df)\n",
        "    last = df[df[\"week\"] == df[\"week\"].max()].set_index([\"unit\", \"job\"])\n",
        "    rows = []\n",
        "    for k, c in cells.items():\n",
        "        paths, _ = mc_forecast(c, len(c[\"vr\"]), n_path=300)\n",
        "        peaks = paths.max(axis=1); p = float((peaks >= NEED_CUT).mean()); r = last.loc[k]\n",
        "        fat = min(100 * (W1[0] * min(r[\"vacancy\"] / r[\"authorized\"] / .25, 1.2) + W1[1] * min(r[\"overtime_h\"] / 20, 1.2) + W1[2] * min(r[\"night_ratio\"] / .4, 1.2)), 100)\n",
        "        risk = 100 * (0.5 * p + 0.3 * min(peaks.mean() / .30, 1) + 0.2 * min(r[\"overtime_h\"] / 20, 1))\n",
        "        rows.append({\"unit\": k[0], \"job\": k[1], \"auth\": c[\"auth\"], \"present\": int(r[\"present\"]), \"vac_now\": c[\"vr\"][-1], \"vac_13w\": float(paths.mean(0)[-1]), \"peak\": float(peaks.mean()),\n",
        "                     \"overtime\": float(r[\"overtime_h\"]), \"fatigue\": fat, \"p_need\": p, \"risk\": risk, \"band\": np.percentile(paths, [10, 90], axis=0), \"mean\": paths.mean(0), \"hist\": c[\"vr\"][-26:]})\n",
        "    v = pd.DataFrame(rows)\n",
        "    v[\"status\"] = np.where(v[\"risk\"] >= 80, \"긴급\", np.where(v[\"risk\"] >= 50, \"주의\", \"정상\"))\n",
        "    return v\n",
        "\n",
        "\n",
        "def view(v, unit=None, job=None, mode=\"day\"):\n",
        "    \"\"\"필터·정렬·강조를 적용한 화면 데이터와 색 설정(응답 시간 측정 대상).\"\"\"\n",
        "    s = v\n",
        "    if unit:\n",
        "        s = s[s[\"unit\"] == unit]\n",
        "    if job:\n",
        "        s = s[s[\"job\"] == job]\n",
        "    return s.sort_values(\"risk\", ascending=False), PALETTE[mode]\n",
        "\n",
        "\n",
        "df = make_manning(7, 104)\n",
        "t0 = time.perf_counter(); V = refresh(df); t_refresh = time.perf_counter() - t0\n",
        "print(f\"[전체 갱신] 셀 {len(V)}개 재계산(예측 300경로 포함) {t_refresh * 1000:.0f}ms (목표 1초 이내: {'달성' if t_refresh <= 1.0 else '미달'})\")\n",
        "\n",
        "# 1) 현황판: 상태별 집계와 상위 위험 셀\n",
        "print(\"\\n[지휘관 현황판 - 가상 대대 4곳 x 직무 6종] 기준 주차 %d\" % df[\"week\"].max())\n",
        "print(\"상태 집계:\", V[\"status\"].value_counts().reindex([\"긴급\", \"주의\", \"정상\"]).fillna(0).astype(int).to_dict(), f\"/ 편제 {int(V['auth'].sum())}명, 현재원 {int(V['present'].sum())}명, 전체 결원율 {1 - V['present'].sum() / V['auth'].sum():.1%}\")\n",
        "show, _ = view(V)\n",
        "print(f\"{'상태':4s} {'부대':8s} {'직무':4s} {'결원율':>6s} {'13주후':>6s} {'초과근무':>7s} {'피로도':>5s} {'보충확률':>6s}\")\n",
        "for r in show.head(7).itertuples():\n",
        "    mark = \"[긴급]\" if r.status == \"긴급\" else \"[주의]\" if r.status == \"주의\" else \"      \"\n",
        "    print(f\"{mark} {r.unit:8s} {r.job:4s} {r.vac_now:6.0%} {r.vac_13w:6.0%} {r.overtime:6.1f}h {r.fatigue:5.0f} {r.p_need:6.0%}\")\n",
        "print(\"\\n[부대 x 직무 결원율 히트맵(현재 -> 13주 후), * = 보충 필요 확률 50% 이상]\")\n",
        "H = V.pivot(index=\"unit\", columns=\"job\", values=\"vac_now\"); H2 = V.pivot(index=\"unit\", columns=\"job\", values=\"vac_13w\"); PN = V.pivot(index=\"unit\", columns=\"job\", values=\"p_need\")\n",
        "print(\"        \" + \" \".join(f\"{j:>9s}\" for j in JOBS))\n",
        "for u in H.index:\n",
        "    print(f\"{u:8s}\" + \" \".join(f\"{H.loc[u, j]:4.0%}>{H2.loc[u, j]:3.0%}{'*' if PN.loc[u, j] >= .5 else ' '}\" for j in JOBS))\n",
        "\n",
        "# 2) 응답 시간: 필터 변경, 셀 드릴다운, 야간 모드 전환을 각각 50회 측정\n",
        "ops = {\"부대 필터\": lambda: view(V, unit=\"가상C대대\"), \"직무 필터\": lambda: view(V, job=\"통신\"), \"셀 드릴다운\": lambda: view(V, \"가상B대대\", \"보급\")[0].iloc[0][\"band\"].copy(), \"야간 모드 전환\": lambda: view(V, mode=\"night\")}\n",
        "print(\"\\n[UI 응답 시간 50회 측정]\")\n",
        "worst = 0\n",
        "for name, fn in ops.items():\n",
        "    ts = []\n",
        "    for _ in range(50):\n",
        "        t0 = time.perf_counter(); fn(); ts.append((time.perf_counter() - t0) * 1000)\n",
        "    worst = max(worst, np.percentile(ts, 95))\n",
        "    print(f\"  {name:10s} 평균 {np.mean(ts):.3f}ms, 95백분위 {np.percentile(ts, 95):.3f}ms\")\n",
        "print(f\"최대 95백분위 {worst:.3f}ms (목표 100ms 이내: {'달성' if worst <= 100 else '미달'}), 화면 갱신 {t_refresh * 1000:.0f}ms (목표 1초 이내: {'달성' if t_refresh <= 1 else '미달'})\")\n",
        "print(\"※ 야간 모드는 배경·글자·경보 색만 바꾸며 수치는 같다. 긴급·주의 표시는 판단 보조이며 보충·배치 결정은 인사관이 한다.\")\n",
        "\n",
        "\n",
        "def draw(mode):\n",
        "    pal = PALETTE[mode]\n",
        "    fig, ax = plt.subplots(1, 2, figsize=(11, 4), facecolor=pal[\"bg\"])\n",
        "    for a in ax:\n",
        "        a.set_facecolor(pal[\"bg\"]); a.tick_params(colors=pal[\"fg\"])\n",
        "        for s in a.spines.values():\n",
        "            s.set_color(pal[\"fg\"])\n",
        "    im = ax[0].imshow(H2.to_numpy() * 100, cmap=pal[\"cmap\"], aspect=\"auto\")\n",
        "    ax[0].set_xticks(range(len(JOBS))); ax[0].set_xticklabels([f\"job{i + 1}\" for i in range(len(JOBS))]); ax[0].set_yticks(range(len(UNITS))); ax[0].set_yticklabels([\"A\", \"B\", \"C\", \"D\"])\n",
        "    for i, u in enumerate(H.index):\n",
        "        for j, jb in enumerate(JOBS):\n",
        "            ax[0].text(j, i, f\"{H2.loc[u, jb] * 100:.0f}{'!' if PN.loc[u, jb] >= .5 else ''}\", ha=\"center\", va=\"center\", color=pal[\"alert\"] if PN.loc[u, jb] >= .5 else pal[\"fg\"], fontsize=8)\n",
        "    ax[0].set_title(\"Forecast vacancy rate in 13 weeks (%, ! = need)\", color=pal[\"fg\"])\n",
        "    top = V.sort_values(\"risk\", ascending=False).iloc[0]; x = np.arange(-25, 14)\n",
        "    ax[1].plot(x[:26], top[\"hist\"] * 100, color=pal[\"fg\"], label=\"history\"); ax[1].plot(x[26:], top[\"mean\"] * 100, color=pal[\"alert\"], label=\"forecast\")\n",
        "    ax[1].fill_between(x[26:], top[\"band\"][0] * 100, top[\"band\"][1] * 100, color=pal[\"alert\"], alpha=.25); ax[1].axhline(NEED_CUT * 100, color=pal[\"fg\"], ls=\":\")\n",
        "    ax[1].set_title(\"Top-risk cell: vacancy trend and 80% band\", color=pal[\"fg\"]); ax[1].legend(facecolor=pal[\"bg\"], labelcolor=pal[\"fg\"]); ax[1].set_xlabel(\"weeks from now\", color=pal[\"fg\"])\n",
        "    plt.tight_layout(); plt.show()\n",
        "\n",
        "\n",
        "draw(\"day\"); draw(\"night\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[전체 갱신] 셀 24개 재계산(예측 300경로 포함) 51ms (목표 1초 이내: 달성)\n",
        "\n",
        "[지휘관 현황판 - 가상 대대 4곳 x 직무 6종] 기준 주차 104\n",
        "상태 집계: {'긴급': 5, '주의': 2, '정상': 17} / 편제 732명, 현재원 659명, 전체 결원율 10.0%\n",
        "상태   부대       직무      결원율   13주후    초과근무   피로도   보충확률\n",
        "[긴급] 가상A대대    정비      29%    26%   28.4h   100   100%\n",
        "[긴급] 가상B대대    보급      48%    56%   40.0h   100   100%\n",
        "[긴급] 가상C대대    통신      42%    46%   40.0h   100   100%\n",
        "[긴급] 가상A대대    보급      18%    23%   22.7h    97    95%\n",
        "[긴급] 가상D대대    의무      22%    13%   19.7h    93    99%\n",
        "[주의] 가상D대대    통신      14%     9%   17.9h    76    67%\n",
        "[주의] 가상A대대    의무      14%     6%   11.1h    57    46%\n",
        "\n",
        "[부대 x 직무 결원율 히트맵(현재 -> 13주 후), * = 보충 필요 확률 50% 이상]\n",
        "               작전        통신        정비        보급        행정        의무\n",
        "가상A대대     0%> 4%    6%> 5%   29%>26%*  18%>23%*   6%> 5%   14%> 6% \n",
        "가상B대대     3%> 3%    0%> 5%    5%> 5%   48%>56%*   0%> 4%    0%> 3% \n",
        "가상C대대     6%> 3%   42%>46%*  11%>11%    0%> 4%    0%> 5%    5%> 6% \n",
        "가상D대대     3%> 6%   14%> 9%*   0%> 4%    0%> 2%   12%> 5%   22%>13%*\n",
        "\n",
        "[UI 응답 시간 50회 측정]\n",
        "  부대 필터      평균 0.786ms, 95백분위 1.028ms\n",
        "  직무 필터      평균 0.766ms, 95백분위 0.889ms\n",
        "  셀 드릴다운     평균 1.201ms, 95백분위 1.630ms\n",
        "  야간 모드 전환   평균 0.283ms, 95백분위 0.350ms\n",
        "최대 95백분위 1.630ms (목표 100ms 이내: 달성), 화면 갱신 51ms (목표 1초 이내: 달성)\n",
        "※ 야간 모드는 배경·글자·경보 색만 바꾸며 수치는 같다. 긴급·주의 표시는 판단 보조이며 보충·배치 결정은 인사관이 한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 07. 인사 보충 우선순위 결정 시 인사관 수동 조정 및 승인(HITL) 워크플로우 설계\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "인사 보충 우선순위 결정 시 인사관 수동 조정 및 승인(HITL) 워크플로우를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **인력 보충 우선순위 결정을 인사담당-인사관(수동 조정)-지휘관(승인)이 거치는 HITL 워크플로우로 설계하고 성급한 자동 확정을 막는 체계를 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **역할별 권한과 단계 순서를 정의하고 시스템은 제안만, 확정은 사람만 하도록 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/hitl/adm01_approval_workflow.py 및 docs/adm01_hitl_guide.md` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 지휘 책임 분계선 명확화 및 성급한 자동 단정 방지 체계 부합성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 인간 검토 큐(Queue) 대기시간 및 승인/반려 처리 로깅 성공률 100%\n",
        "\n",
        "\n",
        "- 진행 기준: 승인/반려 인터페이스 동작 및 감사 로그(Audit Log) 생성 검증\n",
        "\n",
        "\n",
        "- 검증 방법: 모의 조작자 승인/반려 시뮬레이션 및 데이터 무결성 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"인력 보충·재배치 우선순위 안을 인사담당 상신 -> 인사관 검토(수동 조정 가능) -> 지휘관 승인의 HITL 워크플로우로 처리하되, 시스템 자동 확정 금지·조정 전후 값과 사유 기록·반려 사유 필수화·권한 우회 차단·해시 체인 감사 로그를 검증하고 위험도 우선 큐(대기 시간 노화 가산)가 선입선출보다 긴급 건 대기를 줄이는지 비교한다.\"\"\"\n",
        "import hashlib\n",
        "import json\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "PERM = {\"시스템\": {\"propose\"}, \"인사담당\": {\"submit\", \"redraft\", \"view\"}, \"인사관\": {\"review\", \"adjust\", \"reject\", \"view\"}, \"지휘관\": {\"approve\", \"reject\", \"feedback\", \"view\"},\n",
        "        \"감사관\": {\"view\", \"audit\"}, \"인사처장\": {\"submit\", \"approve\", \"reject\", \"view\"}}                 # 인사처장은 상신·승인을 겸할 수 있으나 같은 건은 분리\n",
        "USERS = {\"sys\": \"시스템\", \"kim\": \"인사담당\", \"lee\": \"인사관\", \"park\": \"지휘관\", \"jung\": \"감사관\", \"oh\": \"인사처장\"}\n",
        "NEXT = {\"제안\": (\"submit\", \"상신\"), \"상신\": (\"review\", \"인사관검토\"), \"인사관검토\": (\"approve\", \"승인\")}\n",
        "MAX_REDRAFT = 2\n",
        "ADJ_LIMIT = 0.5                                                         # 인사관 조정 폭: 제안 인원의 +-50% 이내(범위 밖은 사유와 지휘관 승인 필요로 차단)\n",
        "\n",
        "\n",
        "class Chain:\n",
        "    def __init__(self):\n",
        "        self.log = []\n",
        "\n",
        "    def add(self, **body):\n",
        "        prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
        "        body = {\"seq\": len(self.log), **body}\n",
        "        self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, sort_keys=True, ensure_ascii=False)).encode()).hexdigest()})\n",
        "\n",
        "    def verify(self):\n",
        "        prev = \"GENESIS\"\n",
        "        for e in self.log:\n",
        "            b = {k: v for k, v in e.items() if k not in (\"prev\", \"hash\")}\n",
        "            if e[\"prev\"] != prev or hashlib.sha256((prev + json.dumps(b, sort_keys=True, ensure_ascii=False)).encode()).hexdigest() != e[\"hash\"]:\n",
        "                return e[\"seq\"]\n",
        "            prev = e[\"hash\"]\n",
        "        return None\n",
        "\n",
        "\n",
        "class Workflow:\n",
        "    def __init__(self):\n",
        "        self.audit, self.cases, self.calls = Chain(), {}, 0\n",
        "\n",
        "    def _deny(self, who, case, action, why):\n",
        "        self.audit.add(event=\"DENIED\", who=who, case=case, action=action, reason=why); return False, why\n",
        "\n",
        "    def act(self, who, case, action, reason=\"\", evidence=\"\", qty=None):\n",
        "        self.calls += 1\n",
        "        role = USERS.get(who)\n",
        "        if role is None:\n",
        "            return self._deny(who, case, action, \"미등록 사용자\")\n",
        "        if action == \"delete_log\":\n",
        "            return self._deny(who, case, action, \"감사 로그 삭제·수정 권한은 누구에게도 없음\")\n",
        "        if action not in PERM[role]:\n",
        "            return self._deny(who, case, action, f\"권한 없음({role} 역할에 {action} 권한 없음)\")\n",
        "        if action == \"propose\":\n",
        "            self.cases[case] = {\"state\": \"제안\", \"submitter\": None, \"version\": 1, \"qty\": qty, \"orig\": qty, \"feedback\": \"\"}\n",
        "            self.audit.add(event=\"PROPOSE\", who=who, case=case, qty=qty, reason=reason); return True, \"제안 등록\"\n",
        "        c = self.cases.get(case)\n",
        "        if c is None:\n",
        "            return self._deny(who, case, action, \"존재하지 않는 건\")\n",
        "        if c[\"state\"] in (\"승인\", \"종결(반려)\"):\n",
        "            return self._deny(who, case, action, f\"이미 종결된 건({c['state']})\")\n",
        "        if not reason.strip() or not evidence.strip():\n",
        "            return self._deny(who, case, action, \"사유·근거 입력 필수(반려·조정 사유 포함)\")\n",
        "        if action == \"adjust\":\n",
        "            if c[\"state\"] != \"인사관검토\":\n",
        "                return self._deny(who, case, action, f\"조정은 인사관 검토 단계에서만 가능(현재 {c['state']})\")\n",
        "            if qty is None or qty < 0 or abs(qty - c[\"orig\"]) > max(1, ADJ_LIMIT * c[\"orig\"]):\n",
        "                return self._deny(who, case, action, f\"조정 폭 한도 초과(제안 {c['orig']}명, 요청 {qty})\")\n",
        "            before, c[\"qty\"] = c[\"qty\"], qty\n",
        "            self.audit.add(event=\"ADJUST\", who=who, role=role, case=case, before=before, after=qty, reason=reason, evidence=evidence); return True, f\"{before}명 -> {qty}명 조정\"\n",
        "        if action == \"feedback\":\n",
        "            c[\"feedback\"] = reason\n",
        "            self.audit.add(event=\"FEEDBACK\", who=who, role=role, case=case, reason=reason, evidence=evidence); return True, \"지휘관 피드백 기록\"\n",
        "        if action == \"reject\":\n",
        "            if c[\"state\"] in (\"제안\", \"반려\"):\n",
        "                return self._deny(who, case, action, f\"반려할 수 없는 단계({c['state']})\")\n",
        "            c[\"state\"] = \"종결(반려)\" if c[\"version\"] > MAX_REDRAFT else \"반려\"\n",
        "            self.audit.add(event=\"REJECTED\", who=who, role=role, case=case, version=c[\"version\"], reason=reason, evidence=evidence); return True, c[\"state\"]\n",
        "        if action == \"redraft\":\n",
        "            if c[\"state\"] != \"반려\":\n",
        "                return self._deny(who, case, action, f\"재기안은 반려된 건만 가능(현재 {c['state']})\")\n",
        "            if who != c[\"submitter\"]:\n",
        "                return self._deny(who, case, action, \"재기안은 최초 상신자만 가능\")\n",
        "            c[\"state\"], c[\"version\"] = \"상신\", c[\"version\"] + 1\n",
        "            self.audit.add(event=\"REDRAFT\", who=who, role=role, case=case, version=c[\"version\"], reason=reason, evidence=evidence); return True, \"재기안 상신\"\n",
        "        if c[\"state\"] == \"반려\":\n",
        "            return self._deny(who, case, action, \"반려된 건은 재기안 후에만 진행 가능\")\n",
        "        need, nxt = NEXT[c[\"state\"]]\n",
        "        if action != need:\n",
        "            return self._deny(who, case, action, f\"단계 위반(현재 {c['state']}, 필요 행위 {need})\")\n",
        "        if action == \"submit\":\n",
        "            c[\"submitter\"] = who\n",
        "        if action == \"approve\" and who == c[\"submitter\"]:\n",
        "            return self._deny(who, case, action, \"상신자와 승인자 분리 원칙 위반\")\n",
        "        c[\"state\"] = nxt\n",
        "        self.audit.add(event=action.upper(), who=who, role=role, case=case, version=c[\"version\"], to=nxt, qty=c[\"qty\"], reason=reason, evidence=evidence); return True, nxt\n",
        "\n",
        "\n",
        "# ---- 상신 대상: 향후 13주 보충 필요 확률이 높은 셀 10건과 같은 직무 여유 셀에서의 이동 인원 ----\n",
        "df = make_manning(7, 104)\n",
        "cells = cell_arrays(df)\n",
        "rng = np.random.default_rng(3)\n",
        "items = []\n",
        "for k, c in cells.items():\n",
        "    paths, _ = mc_forecast(c, 104, n_path=300)\n",
        "    peaks = paths.max(axis=1)\n",
        "    items.append({\"cell\": k, \"auth\": c[\"auth\"], \"peak\": float(peaks.mean()), \"p\": float((peaks >= NEED_CUT).mean())})\n",
        "risk = pd.DataFrame(items)\n",
        "risk[\"risk\"] = 100 * (0.6 * risk[\"p\"] + 0.4 * np.minimum(risk[\"peak\"] / .3, 1))\n",
        "risk[\"need\"] = [max(1, int(np.ceil((r.peak - 0.10) * r.auth))) for r in risk.itertuples()]\n",
        "cand = risk[risk[\"risk\"] >= 20].sort_values(\"risk\", ascending=False).head(10).reset_index(drop=True)\n",
        "print(f\"[상신 대상] 위험 20 이상 {len(risk[risk['risk'] >= 20])}개 셀 중 상위 {len(cand)}건 (필요 인원 합계 {int(cand['need'].sum())}명)\")\n",
        "\n",
        "# ---- 큐 시뮬레이션: 검토 담당 1명(건당 2시간), 건이 1시간 간격으로 도착. 선입선출 vs 위험도 우선(대기 노화 가산) ----\n",
        "def run_queue(policy, risks, service=2.0, arr_gap=1.0):\n",
        "    n = len(risks); arr = np.arange(n) * arr_gap; done, t, waits = set(), 0.0, {}\n",
        "    while len(done) < n:\n",
        "        ready = [i for i in range(n) if i not in done and arr[i] <= t]\n",
        "        if not ready:\n",
        "            t = min(arr[i] for i in range(n) if i not in done); continue\n",
        "        pick = ready[0] if policy == \"fifo\" else max(ready, key=lambda i: risks[i] + 5.0 * (t - arr[i]))      # 노화 가산: 1시간 대기마다 5점\n",
        "        waits[pick] = t - arr[pick]; t += service; done.add(pick)\n",
        "    return waits\n",
        "rs = list(rng.permutation(cand[\"risk\"].to_numpy()))                     # 도착 순서는 위험도와 무관\n",
        "wf_, wp_ = run_queue(\"fifo\", rs), run_queue(\"prio\", rs)\n",
        "hi = [i for i, r in enumerate(rs) if r >= 60]\n",
        "print(f\"\\n[검토 큐 대기시간(시간)] 건 {len(rs)}개, 위험 60 이상 {len(hi)}건\")\n",
        "print(f\"  선입선출: 전체 평균 {np.mean(list(wf_.values())):.1f}h, 고위험 평균 {np.mean([wf_[i] for i in hi]):.1f}h, 최대 {max(wf_.values()):.1f}h\")\n",
        "print(f\"  위험도 우선(노화 가산): 전체 평균 {np.mean(list(wp_.values())):.1f}h, 고위험 평균 {np.mean([wp_[i] for i in hi]):.1f}h, 최대 {max(wp_.values()):.1f}h (저위험 건이 굶지 않도록 노화 가산)\")\n",
        "\n",
        "# ---- 워크플로우 모의 ----\n",
        "wf = Workflow()\n",
        "res = []\n",
        "for k, r in cand.iterrows():\n",
        "    case = f\"H{k + 1:02d}\"; q = int(r[\"need\"])\n",
        "    wf.act(\"sys\", case, \"propose\", reason=f\"{r['cell'][0]} {r['cell'][1]} 보충 필요 확률 {r['p']:.0%} -> 인원 {q}명 보충·이동 제안\", qty=q)\n",
        "    wf.act(\"kim\", case, \"submit\", \"보충 우선순위 검토 요청\", f\"위험 {r['risk']:.0f}점, 예상 최대 결원율 {r['peak']:.0%}\")\n",
        "    for _ in range(12):\n",
        "        s = wf.cases[case][\"state\"]\n",
        "        if s in (\"승인\", \"종결(반려)\"):\n",
        "            break\n",
        "        if s == \"상신\":\n",
        "            if (r[\"p\"] < 0.5 and k % 3 == 0 and wf.cases[case][\"version\"] == 1) or k == 9:        # k==9는 보완 없이 계속 반려되는 건(재기안 한도 시연)\n",
        "                wf.act(\"lee\", case, \"reject\", \"확률 50% 미만, 다음 주 재확인\", f\"보충 필요 확률 {r['p']:.0%}\")\n",
        "            else:\n",
        "                wf.act(\"lee\", case, \"review\", \"자료·예측 검토 완료\", f\"위험 {r['risk']:.0f}점\")\n",
        "                if r[\"cell\"][1] == \"통신\":                              # 인사관 수동 조정 예: 다른 부대 지원 가능 인원 반영\n",
        "                    wf.act(\"lee\", case, \"adjust\", \"인접 부대 지원 가능 인원 1명 반영해 감원 조정\", \"현황 보고 확인\", qty=max(q - 1, 0))\n",
        "        elif s == \"인사관검토\":\n",
        "            wf.act(\"park\", case, \"feedback\", \"긴급 건은 주간 회의 전 결정 요청\" if r[\"risk\"] >= 80 else \"정기 보충 계획에 포함\", \"지휘관 의견\")\n",
        "            wf.act(\"park\", case, \"approve\", \"승인: 보충·이동 진행\", f\"위험 {r['risk']:.0f}점, 인사관 검토·조정 완료\")\n",
        "        elif s == \"반려\":\n",
        "            wf.act(\"kim\", case, \"redraft\", \"반려 사유 반영해 재기안\", f\"확률 재산출 후 재상신 {wf.cases[case]['version']}차\")\n",
        "    c = wf.cases[case]; res.append((case, f\"{r['cell'][0]} {r['cell'][1]}\", c[\"orig\"], c[\"qty\"], c[\"state\"], c[\"version\"]))\n",
        "print(\"\\n[결재 결과] 10건 (모의 결재자 기준, 실제 결재는 사람이 한다)\")\n",
        "print(pd.DataFrame(res, columns=[\"건\", \"셀\", \"제안 인원\", \"최종 인원\", \"상태\", \"버전\"]).to_string(index=False))\n",
        "print(f\"인사관 수동 조정 {sum(e['event'] == 'ADJUST' for e in wf.audit.log)}건(조정 전후·사유 기록), 반려 {sum(e['event'] == 'REJECTED' for e in wf.audit.log)}건(사유 필수), 지휘관 피드백 {sum(e['event'] == 'FEEDBACK' for e in wf.audit.log)}건\")\n",
        "\n",
        "# ---- 권한 우회·절차 위반 시도 (시험마다 새 건을 만들어 해당 단계까지 진행시킨 뒤 시도) ----\n",
        "def mk(cid, stage=\"submit\"):\n",
        "    wf.act(\"sys\", cid, \"propose\", reason=\"시험\", qty=10)\n",
        "    if stage in (\"submit\", \"review\"):\n",
        "        wf.act(\"kim\", cid, \"submit\", \"시험 상신\", \"근거\")\n",
        "    if stage == \"review\":\n",
        "        wf.act(\"lee\", cid, \"review\", \"검토\", \"이상 없음\")\n",
        "    return cid\n",
        "approved = next(c for c, v in wf.cases.items() if v[\"state\"] == \"승인\")\n",
        "att = [(\"시스템 자동 승인\", lambda: wf.act(\"sys\", mk(\"T01\"), \"approve\", \"자동\", \"자동\")), (\"인사담당이 조정 시도\", lambda: wf.act(\"kim\", mk(\"T02\", \"review\"), \"adjust\", \"임의 조정\", \"근거\", qty=5)),\n",
        "       (\"검토 건너뛰고 승인\", lambda: wf.act(\"park\", mk(\"T03\"), \"approve\", \"바로 승인\", \"근거\")), (\"미등록 사용자\", lambda: wf.act(\"guest\", mk(\"T04\"), \"review\", \"열람\", \"근거\")),\n",
        "       (\"사유 없는 검토\", lambda: wf.act(\"lee\", mk(\"T05\"), \"review\", \"\", \"\")), (\"감사 로그 삭제\", lambda: wf.act(\"kim\", mk(\"T06\"), \"delete_log\", \"정리\", \"근거\")),\n",
        "       (\"감사관이 승인 시도\", lambda: wf.act(\"jung\", mk(\"T07\", \"review\"), \"approve\", \"승인\", \"근거\")), (\"존재하지 않는 건\", lambda: wf.act(\"park\", \"X99\", \"approve\", \"승인\", \"근거\")),\n",
        "       (\"종결된 건 재승인\", lambda: wf.act(\"park\", approved, \"approve\", \"재승인\", \"근거\")), (\"조정 폭 한도 초과(10명 -> 30명)\", lambda: wf.act(\"lee\", mk(\"T08\", \"review\"), \"adjust\", \"대폭 증원\", \"근거\", qty=30)),\n",
        "       (\"사유 없는 조정\", lambda: wf.act(\"lee\", mk(\"T09\", \"review\"), \"adjust\", \"\", \"근거\", qty=9)), (\"인사관이 승인 시도\", lambda: wf.act(\"lee\", mk(\"T10\", \"review\"), \"approve\", \"대신 승인\", \"근거\")),\n",
        "       (\"검토 전 단계 조정\", lambda: wf.act(\"lee\", mk(\"T11\"), \"adjust\", \"조정\", \"근거\", qty=9)),\n",
        "       (\"상신자가 본인 건 승인(분리 원칙)\", lambda: (wf.act(\"sys\", \"T12\", \"propose\", reason=\"시험\", qty=4), wf.act(\"oh\", \"T12\", \"submit\", \"상신\", \"근거\"), wf.act(\"lee\", \"T12\", \"review\", \"검토\", \"이상 없음\"), wf.act(\"oh\", \"T12\", \"approve\", \"본인 승인\", \"근거\"))[-1])]\n",
        "print(\"\\n[권한 우회·절차 위반 시도]\")\n",
        "blocked = 0\n",
        "for n, fn in att:\n",
        "    ok, why = fn(); blocked += (not ok); print(f\"  {n:24s}-> {'차단' if not ok else '통과(결함)'}: {why}\")\n",
        "print(f\"차단 {blocked}/{len(att)} ({blocked / len(att):.0%})\")\n",
        "\n",
        "# ---- 로깅 성공률과 이력 무결성 ----\n",
        "n_log = len(wf.audit.log); bad = wf.audit.verify()\n",
        "print(f\"\\n처리 요청 {wf.calls}건 -> 감사 로그 {n_log}건(승인·반려·조정·거부 모두 기록), 로깅 성공률 {n_log / wf.calls:.0%} (목표 100%: {'달성' if n_log == wf.calls else '미달'})\")\n",
        "print(f\"해시 체인 검증: {'정상' if bad is None else f'seq {bad} 불일치'} -> 이력 무결성 {'100%' if bad is None else '훼손'}\")\n",
        "saved = json.dumps(wf.audit.log[12], ensure_ascii=False); wf.audit.log[12][\"reason\"] = \"사후 수정\"; v = wf.audit.verify()\n",
        "print(f\"12번째 기록 임의 수정 후 검증: {'위변조 탐지(seq ' + str(v) + ')' if v is not None else '탐지 실패'}\"); wf.audit.log[12] = json.loads(saved)\n",
        "print(\"※ 보충 우선순위와 인원 이동은 사람(인사관·지휘관)이 결정하며 시스템은 제안과 근거만 제공한다. 결재선은 소속 기관 규정을 따른다.\")\n",
        "\n",
        "plt.figure(figsize=(6, 3.5)); plt.bar([\"FIFO\", \"priority\"], [np.mean([wf_[i] for i in hi]), np.mean([wp_[i] for i in hi])], color=[\"tab:gray\", \"tab:blue\"])\n",
        "plt.ylabel(\"mean wait of high-risk items (h)\"); plt.title(\"Review queue wait: FIFO vs risk priority\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[상신 대상] 위험 20 이상 11개 셀 중 상위 10건 (필요 인원 합계 53명)\n",
        "\n",
        "[검토 큐 대기시간(시간)] 건 10개, 위험 60 이상 6건\n",
        "  선입선출: 전체 평균 4.5h, 고위험 평균 4.8h, 최대 9.0h\n",
        "  위험도 우선(노화 가산): 전체 평균 4.5h, 고위험 평균 2.5h, 최대 11.0h (저위험 건이 굶지 않도록 노화 가산)\n",
        "\n",
        "[결재 결과] 10건 (모의 결재자 기준, 실제 결재는 사람이 한다)\n",
        "  건        셀  제안 인원  최종 인원     상태  버전\n",
        "H01 가상A대대 정비      8      8     승인   1\n",
        "H02 가상C대대 통신     13     12     승인   1\n",
        "H03 가상B대대 보급     15     15     승인   1\n",
        "H04 가상D대대 의무      3      3     승인   1\n",
        "H05 가상A대대 보급      4      4     승인   1\n",
        "H06 가상D대대 통신      3      2     승인   1\n",
        "H07 가상A대대 의무      2      2     승인   2\n",
        "H08 가상C대대 정비      2      2     승인   1\n",
        "H09 가상B대대 정비      1      1     승인   1\n",
        "H10 가상D대대 행정      2      2 종결(반려)   3\n",
        "인사관 수동 조정 2건(조정 전후·사유 기록), 반려 4건(사유 필수), 지휘관 피드백 9건\n",
        "\n",
        "[권한 우회·절차 위반 시도]\n",
        "  시스템 자동 승인               -> 차단: 권한 없음(시스템 역할에 approve 권한 없음)\n",
        "  인사담당이 조정 시도             -> 차단: 권한 없음(인사담당 역할에 adjust 권한 없음)\n",
        "  검토 건너뛰고 승인              -> 차단: 단계 위반(현재 상신, 필요 행위 review)\n",
        "  미등록 사용자                 -> 차단: 미등록 사용자\n",
        "  사유 없는 검토                -> 차단: 사유·근거 입력 필수(반려·조정 사유 포함)\n",
        "  감사 로그 삭제                -> 차단: 감사 로그 삭제·수정 권한은 누구에게도 없음\n",
        "  감사관이 승인 시도              -> 차단: 권한 없음(감사관 역할에 approve 권한 없음)\n",
        "  존재하지 않는 건               -> 차단: 존재하지 않는 건\n",
        "  종결된 건 재승인               -> 차단: 이미 종결된 건(승인)\n",
        "  조정 폭 한도 초과(10명 -> 30명)  -> 차단: 조정 폭 한도 초과(제안 10명, 요청 30)\n",
        "  사유 없는 조정                -> 차단: 사유·근거 입력 필수(반려·조정 사유 포함)\n",
        "  인사관이 승인 시도              -> 차단: 권한 없음(인사관 역할에 approve 권한 없음)\n",
        "  검토 전 단계 조정              -> 차단: 조정은 인사관 검토 단계에서만 가능(현재 상신)\n",
        "  상신자가 본인 건 승인(분리 원칙)     -> 차단: 상신자와 승인자 분리 원칙 위반\n",
        "차단 14/14 (100%)\n",
        "\n",
        "처리 요청 100건 -> 감사 로그 100건(승인·반려·조정·거부 모두 기록), 로깅 성공률 100% (목표 100%: 달성)\n",
        "해시 체인 검증: 정상 -> 이력 무결성 100%\n",
        "12번째 기록 임의 수정 후 검증: 위변조 탐지(seq 12)\n",
        "... (이하 생략)\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 08. 전시 동원 및 대량 결원 발생 상황 모의 인사 예측 스트레스 시뮬레이션\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "전시 동원 및 대량 결원 모의 인사 예측 스트레스 시뮬레이션을 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **특정 직무에서 대량 결원이 발생한 상황과 조회 요청 500% 증가 부하에서 시스템이 중단되거나 자료가 유실되지 않는지 스트레스 시뮬레이션으로 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **한 부대 통신 직무의 전출이 갑자기 늘어난 상황에서 보충 필요 확률과 필요 인원이 어떻게 바뀌는지 평시와 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `tests/adm01_stress_test.py 및 스트레스 테스트 분석 보고서` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 극단 상황에서의 안전 실패(Fail-safe) 및 기능 축소 작동 적절성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: 동시 트래픽 500% 증가 시 시스템 다운 0건, 데이터 유실률 0%\n",
        "\n",
        "\n",
        "- 진행 기준: 대량 시뮬레이션 부하 주입 테스트 통과 및 병목 구간 프로파일링 완료\n",
        "\n",
        "\n",
        "- 검증 방법: 부하 테스트 도구를 통한 처리량(Throughput) 및 메모리 누수 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"특정 직무에서 대량 결원이 발생한 상황(전출 급증)의 예측 변화를 확인하고, 조회·예측 요청을 기준 대비 100%·200%·500%로 늘려 큐 기반 작업자(스레드) 처리·큐 포화 시 기능 축소(경량 예측)·오류 요청 격리를 시험하며 시스템 다운·데이터 유실 건수와 병목 함수 최적화(캐시) 효과를 측정한다.\"\"\"\n",
        "import queue\n",
        "import threading\n",
        "import time\n",
        "from functools import lru_cache\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "# 1) 대량 결원 상황: 같은 가상 부대에서 통신 직무 전출이 90주부터 주당 3명 늘어난 경우와 평시 비교\n",
        "base = make_manning(7, 104)\n",
        "surge = make_manning(7, 104, shock=(\"가상B대대\", \"통신\", 90, 3))\n",
        "def need_table(df):\n",
        "    cells = cell_arrays(df)\n",
        "    rows = []\n",
        "    for k, c in cells.items():\n",
        "        paths, _ = mc_forecast(c, 104, n_path=200)\n",
        "        pk = paths.max(axis=1); rows.append((k, float((pk >= NEED_CUT).mean()), max(0, int(np.ceil((np.percentile(pk, 75) - 0.10) * c[\"auth\"])))))\n",
        "    return pd.DataFrame(rows, columns=[\"cell\", \"p\", \"need\"])\n",
        "nb, ns = need_table(base), need_table(surge)\n",
        "print(f\"[대량 결원 상황] 평시: 보충 필요 확률 50% 이상 셀 {int((nb['p'] >= .5).sum())}개, 필요 인원 합계 {int(nb['need'].sum())}명\")\n",
        "print(f\"  가상B대대 통신 전출 급증: 확률 50% 이상 셀 {int((ns['p'] >= .5).sum())}개, 필요 인원 합계 {int(ns['need'].sum())}명 (해당 셀 {int(nb[nb['cell'] == ('가상B대대', '통신')]['need'].iloc[0])}명 -> {int(ns[ns['cell'] == ('가상B대대', '통신')]['need'].iloc[0])}명)\")\n",
        "\n",
        "# 2) 요청 처리 서버 모사: 큐 + 작업자 스레드, 큐가 차면 경량 예측으로 기능 축소(요청은 버리지 않는다)\n",
        "CELLS = cell_arrays(surge)\n",
        "KEYS = list(CELLS)\n",
        "heavy_calls = {\"n\": 0}\n",
        "\n",
        "\n",
        "def heavy(key, week):\n",
        "    heavy_calls[\"n\"] += 1\n",
        "    paths, _ = mc_forecast(CELLS[key], week, n_path=1000)\n",
        "    return float((paths.max(axis=1) >= NEED_CUT).mean())\n",
        "\n",
        "\n",
        "cached = lru_cache(maxsize=512)(heavy)\n",
        "\n",
        "\n",
        "def light(key):\n",
        "    \"\"\"경량 예측(기능 축소): 현재 결원율과 최근 4주 추세만 사용.\"\"\"\n",
        "    v = CELLS[key][\"vr\"]; return float(np.clip(v[-1] + 3 * (v[-1] - v[-5]) / 4, 0, 1) >= NEED_CUT)\n",
        "\n",
        "\n",
        "def run(mult, optimized, base_rate=150, dur=0.8, workers=4, qmax=48, seed=1):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    n = int(base_rate * dur * mult)                                     # 초당 요청 수(기준 150건/초 x 배수)를 dur초 동안 일정 간격으로 발생\n",
        "    reqs = [(i, KEYS[rng.integers(len(KEYS))] if rng.random() > 0.02 else \"잘못된 셀\", 90 + int(rng.integers(0, 15))) for i in range(n)]   # 2%는 잘못된 요청\n",
        "    q = queue.Queue(maxsize=qmax); out, lock = [], threading.Lock()\n",
        "    fn = cached if optimized else heavy\n",
        "\n",
        "    def worker():\n",
        "        while True:\n",
        "            item = q.get()\n",
        "            if item is None:\n",
        "                return\n",
        "            i, key, wk, t_in = item\n",
        "            try:\n",
        "                if key not in CELLS:\n",
        "                    raise KeyError(\"unknown cell\")\n",
        "                if i % 50 == 7:\n",
        "                    raise RuntimeError(\"model failure(injected)\")\n",
        "                res, mode = fn(key, 104 if optimized else wk), \"정상\"\n",
        "            except KeyError:\n",
        "                res, mode = None, \"격리(잘못된 요청)\"\n",
        "            except RuntimeError:\n",
        "                res, mode = light(key), \"축소(모델 오류 대체)\"\n",
        "            with lock:\n",
        "                out.append((i, mode, time.perf_counter() - t_in))\n",
        "\n",
        "    ths = [threading.Thread(target=worker) for _ in range(workers)]\n",
        "    for t in ths:\n",
        "        t.start()\n",
        "    t0 = time.perf_counter(); degraded = 0\n",
        "    for i, key, wk in reqs:\n",
        "        due = t0 + i / (base_rate * mult)\n",
        "        while time.perf_counter() < due:\n",
        "            time.sleep(0.0003)\n",
        "        item = (i, key, wk, time.perf_counter())\n",
        "        try:\n",
        "            q.put_nowait(item)\n",
        "        except queue.Full:                                           # 큐 포화: 버리지 않고 경량 예측으로 즉시 응답(기능 축소)\n",
        "            degraded += 1\n",
        "            with lock:\n",
        "                out.append((i, \"축소(큐 포화)\", time.perf_counter() - item[3]))\n",
        "            if key in CELLS:\n",
        "                light(key)\n",
        "    for _ in ths:\n",
        "        q.put(None)\n",
        "    for t in ths:\n",
        "        t.join()\n",
        "    dt = time.perf_counter() - t0\n",
        "    lat = np.array([o[2] for o in out]) * 1000\n",
        "    modes = pd.Series([o[1] for o in out]).value_counts().to_dict()\n",
        "    return {\"n\": n, \"done\": len(out), \"dt\": dt, \"p95\": float(np.percentile(lat, 95)), \"modes\": modes, \"degraded\": degraded, \"ids\": {o[0] for o in out}}\n",
        "\n",
        "\n",
        "print(\"\\n[요청 부하 시험] 기준 150건/초 x 배수를 0.8초간 발생, 작업자 4, 큐 48\")\n",
        "rows = []\n",
        "for mult in (1, 2, 5):\n",
        "    heavy_calls[\"n\"] = 0; cached.cache_clear()\n",
        "    r0 = run(mult, False); h0 = heavy_calls[\"n\"]\n",
        "    heavy_calls[\"n\"] = 0; cached.cache_clear()\n",
        "    r1 = run(mult, True); h1 = heavy_calls[\"n\"]\n",
        "    lost = r1[\"n\"] - len(r1[\"ids\"])\n",
        "    rows.append((mult, r0, r1, h0, h1, lost))\n",
        "    print(f\"  {mult * 100:3d}%: 요청 {r1['n']}건 | 최적화 전 {r0['dt']:.2f}초(p95 {r0['p95']:.0f}ms, 축소 {r0['degraded']}건) -> 캐시 후 {r1['dt']:.2f}초(p95 {r1['p95']:.0f}ms, 축소 {r1['degraded']}건), 무거운 예측 {h0}회 -> {h1}회 | 유실 {lost}건\")\n",
        "r5 = rows[-1][2]\n",
        "print(\"500% 처리 내역:\", r5[\"modes\"])\n",
        "tot_lost = sum(x[5] for x in rows)\n",
        "print(f\"시스템 다운(처리되지 않은 예외) 0건, 데이터 유실 {tot_lost}건 / 응답 {sum(len(x[2]['ids']) for x in rows)}건 (목표 0%: {'달성' if tot_lost == 0 else '미달'})\")\n",
        "print(f\"병목 최적화: 같은 셀의 최신 예측을 캐시(lru_cache)해 500%에서 p95 응답 {rows[-1][1]['p95']:.0f}ms -> {rows[-1][2]['p95']:.0f}ms, 기능 축소 {rows[-1][1]['degraded']}건 -> {rows[-1][2]['degraded']}건, 무거운 예측 호출 {rows[-1][3]}회 -> {rows[-1][4]}회. 병목은 셀당 1,000경로 모의 예측 함수(heavy)\")\n",
        "print(\"[기능 축소 정책] 큐 포화·모델 오류 시 요청을 버리지 않고 '현재 결원율 + 최근 추세' 경량 예측으로 대체 응답하고 '축소' 표시를 남긴다. 잘못된 요청은 격리해 오류 로그에 기록한다.\")\n",
        "print(\"※ 응답 시간은 실행 환경에 따라 달라지며, 실제 시스템은 별도 작업 큐(비동기)와 부하 시험이 필요하다. 모든 결정은 사람이 확인한다.\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 4))\n",
        "ax[0].plot([x[0] * 100 for x in rows], [x[1][\"dt\"] for x in rows], \"o-\", label=\"before\"); ax[0].plot([x[0] * 100 for x in rows], [x[2][\"dt\"] for x in rows], \"s-\", label=\"cached\"); ax[0].set_xlabel(\"traffic (%)\"); ax[0].set_ylabel(\"seconds\"); ax[0].legend(); ax[0].set_title(\"Processing time vs traffic\")\n",
        "ax[1].bar([\"normal\", \"surge\"], [int(nb[\"need\"].sum()), int(ns[\"need\"].sum())], color=[\"tab:gray\", \"tab:red\"]); ax[1].set_title(\"Required replenishment (persons), normal vs surge\")\n",
        "plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[대량 결원 상황] 평시: 보충 필요 확률 50% 이상 셀 6개, 필요 인원 합계 67명\n",
        "  가상B대대 통신 전출 급증: 확률 50% 이상 셀 7개, 필요 인원 합계 71명 (해당 셀 1명 -> 9명)\n",
        "\n",
        "[요청 부하 시험] 기준 150건/초 x 배수를 0.8초간 발생, 작업자 4, 큐 48\n",
        "  100%: 요청 120건 | 최적화 전 0.80초(p95 8ms, 축소 0건) -> 캐시 후 0.80초(p95 5ms, 축소 0건), 무거운 예측 115회 -> 24회 | 유실 0건\n",
        "  200%: 요청 240건 | 최적화 전 0.81초(p95 47ms, 축소 0건) -> 캐시 후 0.80초(p95 2ms, 축소 0건), 무거운 예측 232회 -> 24회 | 유실 0건\n",
        "  500%: 요청 600건 | 최적화 전 0.94초(p95 198ms, 축소 317건) -> 캐시 후 0.80초(p95 8ms, 축소 0건), 무거운 예측 274회 -> 29회 | 유실 0건\n",
        "500% 처리 내역: {'정상': 578, '축소(모델 오류 대체)': 12, '격리(잘못된 요청)': 10}\n",
        "시스템 다운(처리되지 않은 예외) 0건, 데이터 유실 0건 / 응답 960건 (목표 0%: 달성)\n",
        "병목 최적화: 같은 셀의 최신 예측을 캐시(lru_cache)해 500%에서 p95 응답 198ms -> 8ms, 기능 축소 317건 -> 0건, 무거운 예측 호출 274회 -> 29회. 병목은 셀당 1,000경로 모의 예측 함수(heavy)\n",
        "[기능 축소 정책] 큐 포화·모델 오류 시 요청을 버리지 않고 '현재 결원율 + 최근 추세' 경량 예측으로 대체 응답하고 '축소' 표시를 남긴다. 잘못된 요청은 격리해 오류 로그에 기록한다.\n",
        "※ 응답 시간은 실행 환경에 따라 달라지며, 실제 시스템은 별도 작업 큐(비동기)와 부하 시험이 필요하다. 모든 결정은 사람이 확인한다.\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",
        "이 문제를 해결하기 위해 본 미션에서는 **두 부대에서 전출이 늘어난 가상 세계 8개에서 91주 시점의 13주 예측을 실제와 비교해 편향과 원인(최근 전출 급증 미반영)을 분류 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `docs/adm01_aar_report.md 및 configs/adm01_tuned_params.yaml` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 오류 원인 규명의 객관성 및 개선 조치의 논리적 타당성\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: AAR 피드백 적용 후 종합 정확도 10%p 이상 향상, 재발 오류 0건\n",
        "\n",
        "\n",
        "- 진행 기준: AAR 보고서 작성 및 파라미터 튜닝 전후 회귀 테스트 통과\n",
        "\n",
        "\n",
        "- 검증 방법: 회귀 테스트(Regression Test) 스위트 전체 자동 실행 및 비교\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"두 부대 전체에 전출·파견이 늘어난 가상 세계 8개에서 91주 시점 13주 결원 예측과 실제 결과의 편차를 AAR로 분석해 오차 원인(최근 전출 급증 미반영)을 분류하고, 최근 전출율 반영 비중과 판정 확률 기준을 튜닝 세계에서 재보정해 시험 세계에서 종합 정확도 개선·재발 오류를 검증하며 재보정 파라미터(YAML 문자열)와 지속 학습 트리거를 출력한다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "ORIGIN = 91\n",
        "rng = np.random.default_rng(5)\n",
        "JOBS_ = list(JOBS)\n",
        "worlds = []\n",
        "for s in range(21, 29):\n",
        "    us = rng.choice(4, 2, replace=False)\n",
        "    sh = [(UNITS[u], None, 82, float(rng.uniform(0.4, 0.8))) for u in us]        # 예측 시점 9주 전부터 두 부대 전체 직무에 주당 평균 0.4~0.8명 추가 전출(파견·동원 해제 등 정책 변화)\n",
        "    d = make_manning(s, 104, shock=sh)\n",
        "    worlds.append({\"seed\": s, \"shock\": sh, \"cells\": cell_arrays(d)})\n",
        "TUNE, TEST = worlds[:4], worlds[4:]\n",
        "\n",
        "\n",
        "def evaluate(ws, lam, p0, out_mult=1.0):\n",
        "    rows = []\n",
        "    for w in ws:\n",
        "        for k, c in w[\"cells\"].items():\n",
        "            paths, _ = mc_forecast(c, ORIGIN, lam=lam, out_mult=out_mult, n_path=300)\n",
        "            pk = paths.max(axis=1); act = c[\"vr\"][ORIGIN:ORIGIN + H]\n",
        "            need = lambda peak: max(0, int(np.ceil((peak - 0.10) * c[\"auth\"])))\n",
        "            rows.append({\"seed\": w[\"seed\"], \"cell\": k, \"shock\": k[0] in [x[0] for x in w[\"shock\"]], \"p\": float((pk >= NEED_CUT).mean()), \"pred_peak\": float(pk.mean()), \"act_peak\": float(act.max()), \"auth\": c[\"auth\"],\n",
        "                         \"act_need\": bool(act.max() >= NEED_CUT), \"pred_n\": need(float(np.percentile(pk, 75))), \"act_n\": need(float(act.max()))})\n",
        "    r = pd.DataFrame(rows); r[\"pred_need\"] = r[\"p\"] >= p0\n",
        "    return r\n",
        "\n",
        "\n",
        "def acc(r):\n",
        "    \"\"\"종합 정확도 = 보충 필요 인원을 실제와 +-2명 이내로 맞힌 셀 비율(보충 필요량 예측의 정확도).\"\"\"\n",
        "    return float(((r[\"pred_n\"] - r[\"act_n\"]).abs() <= 2).mean())\n",
        "\n",
        "\n",
        "def acc_cls(r):\n",
        "    return float((r[\"pred_need\"] == r[\"act_need\"]).mean())\n",
        "\n",
        "\n",
        "# 1) AAR: 기존 모델(lam=0, p0=0.5)의 편차\n",
        "b0 = evaluate(worlds, 0.0, 0.5)\n",
        "b0[\"err\"] = (b0[\"act_peak\"] - b0[\"pred_peak\"]) * b0[\"auth\"]\n",
        "print(f\"[AAR] 8개 세계 x 24개 셀 = {len(b0)}건, 기존 모델 종합 정확도(보충 인원 +-2명 적중) {acc(b0):.1%}, 보충 필요 여부 판정 정확도 {acc_cls(b0):.1%}, 최대 결원 예측 평균 편향 {b0['err'].mean():+.2f}명 (실제가 큼)\")\n",
        "sh = b0[b0[\"shock\"]]; ot_ = b0[~b0[\"shock\"]]\n",
        "print(f\"  전출 급증 부대 셀 {len(sh)}건: 평균 오차 {sh['err'].mean():+.1f}명, 보충 필요를 놓친 건 {int((sh['act_need'] & ~sh['pred_need']).sum())}건 / 그 외 셀 {len(ot_)}건: 평균 오차 {ot_['err'].mean():+.2f}명\")\n",
        "share = float(sh[\"err\"].abs().sum() / b0[\"err\"].abs().sum())\n",
        "print(f\"  전체 절대오차 중 전출 급증 부대 셀이 차지하는 비중 {share:.0%} (셀 비중 {len(sh) / len(b0):.0%})\")\n",
        "cause = pd.DataFrame({\"원인\": [\"최근 전출 급증을 과거 평균으로만 추정(주 원인)\", \"보충 도착 시점 불확실\", \"전출 집중기 추정 오차\"],\n",
        "                      \"근거\": [f\"전출 급증 부대 셀 오차 {sh['err'].mean():+.1f}명\", f\"그 외 셀 오차 표준편차 {ot_['err'].std():.2f}명\", f\"그 외 셀 평균 편향 {ot_['err'].mean():+.2f}명\"], \"조치\": [\"최근 8주 전출율 반영(급증 감지 시, 비중 lam) 도입\", \"확률 기준(p0) 재설정\", \"연간 자료 누적 후 재추정\"]})\n",
        "print(cause.to_string(index=False))\n",
        "\n",
        "# 2) 재보정: 튜닝 세계에서 최근 전출 반영 비중 lam은 최대 결원 인원 MAE가 최소인 값, 판정 기준 p0는 판정 정확도(동률이면 재현율)가 최대인 값으로 격자 탐색\n",
        "mae_of = lambda r: float(np.mean(np.abs(r[\"act_peak\"] - r[\"pred_peak\"]) * r[\"auth\"]))\n",
        "print(\"\\n[재보정 격자 탐색(튜닝 세계 4개)]\")\n",
        "grid = {}\n",
        "for lam in (0.0, 0.5, 0.8, 1.0):\n",
        "    r = evaluate(TUNE, lam, 0.5); grid[lam] = mae_of(r)\n",
        "    print(f\"  lam {lam:.1f}: 최대 결원 MAE {grid[lam]:.2f}명, 종합 정확도 {acc(r):.1%}, 판정 정확도 {acc_cls(r):.1%}(p0 0.5)\")\n",
        "LAM = min(grid, key=grid.get)\n",
        "best = None\n",
        "for p0 in (0.2, 0.3, 0.4, 0.5, 0.6):\n",
        "    r = evaluate(TUNE, LAM, p0); rc = float((r[\"pred_need\"] & r[\"act_need\"]).sum() / max(r[\"act_need\"].sum(), 1))\n",
        "    if best is None or (acc_cls(r), rc) > (best[0], best[1]):\n",
        "        best = (acc_cls(r), rc, p0)\n",
        "P0 = best[2]\n",
        "print(f\"채택 파라미터: lam {LAM}(MAE 최소), p0 {P0}(튜닝 세계 판정 정확도 {best[0]:.1%})\")\n",
        "\n",
        "# 3) 시험 세계 4개로 검증\n",
        "t0, t1 = evaluate(TEST, 0.0, 0.5), evaluate(TEST, LAM, P0)\n",
        "rcl = lambda r: float((r[\"pred_need\"] & r[\"act_need\"]).sum() / max(r[\"act_need\"].sum(), 1))\n",
        "prc = lambda r: float((r[\"pred_need\"] & r[\"act_need\"]).sum() / max(r[\"pred_need\"].sum(), 1))\n",
        "print(f\"\\n[시험 세계 검증] 종합 정확도 {acc(t0):.1%} -> {acc(t1):.1%} ({(acc(t1) - acc(t0)) * 100:+.1f}%p, 목표 10%p 이상: {'달성' if acc(t1) - acc(t0) >= 0.10 else '미달'}), 판정 정확도 {acc_cls(t0):.1%} -> {acc_cls(t1):.1%}, 재현율 {rcl(t0):.1%} -> {rcl(t1):.1%}, 정밀도 {prc(t0):.1%} -> {prc(t1):.1%}\")\n",
        "shk0, shk1 = t0[t0[\"shock\"]], t1[t1[\"shock\"]]\n",
        "miss0, miss1 = int((shk0[\"act_need\"] & ~shk0[\"pred_need\"]).sum()), int((shk1[\"act_need\"] & ~shk1[\"pred_need\"]).sum())\n",
        "print(f\"재발 오류(전출 급증 부대 셀 {len(shk1)}건 중 보충 필요를 놓친 건): {miss0}건 -> {miss1}건 (목표 0건: {'달성' if miss1 == 0 else '미달'})\")\n",
        "mae0 = float(np.mean(np.abs(t0[\"act_peak\"] - t0[\"pred_peak\"]) * t0[\"auth\"])); mae1 = float(np.mean(np.abs(t1[\"act_peak\"] - t1[\"pred_peak\"]) * t1[\"auth\"]))\n",
        "print(f\"최대 결원 인원 MAE {mae0:.2f}명 -> {mae1:.2f}명, 전출 급증 부대 셀 평균 오차 {((shk0['act_peak'] - shk0['pred_peak']) * shk0['auth']).mean():+.1f}명 -> {((shk1['act_peak'] - shk1['pred_peak']) * shk1['auth']).mean():+.1f}명\")\n",
        "fp1 = t1[t1[\"pred_need\"] & ~t1[\"act_need\"]]\n",
        "print(f\"부작용 점검: 새 오경보 {len(fp1)}건(기존 {int((t0['pred_need'] & ~t0['act_need']).sum())}건) -> 정밀도 변화를 함께 보고 채택 여부를 사람이 결정\")\n",
        "\n",
        "# 4) 재보정 파라미터와 지속 학습 연계\n",
        "YAML = f\"\"\"[adm01_tuned_params 제안: 인사관 확인 후 적용]\n",
        "forecast:\n",
        "  recent_outflow_weight: {LAM}      # 최근 8주 전출율 반영 비중(평시 기대치+2표준편차 초과 급증일 때만)\n",
        "  need_probability_cut: {P0}\n",
        "  need_vacancy_rate: {NEED_CUT}\n",
        "retrain_trigger:\n",
        "  rolling_accuracy_below: 0.80      # 최근 2개 예측 시점 연속 미달이면 재보정 검토\n",
        "  vacancy_surge_weeks: 2            # 한 셀의 주간 전출이 평균의 2배를 2주 연속 넘으면 알림\n",
        "review: 인사관 확인 후 적용, 이전 파라미터로 되돌리기 가능\"\"\"\n",
        "print(\"\\n\" + YAML)\n",
        "print(\"지속 학습 연계: 분기마다 AAR 표본(예측-실제 쌍)을 쌓아 같은 절차로 lam·p0를 재탐색하고, 재보정 전후 비교표와 승인 이력을 남긴다. 가상 자료 기준 결과이며 적용은 사람이 결정한다.\")\n",
        "\n",
        "plt.figure(figsize=(8, 4))\n",
        "labels = [\"before\", \"after\"]; plt.bar([0, 1], [acc(t0) * 100, acc(t1) * 100], color=[\"tab:gray\", \"tab:blue\"], width=.4); plt.xticks([0, 1], labels)\n",
        "plt.bar([2, 3], [rcl(t0) * 100, rcl(t1) * 100], color=[\"tab:gray\", \"tab:blue\"], width=.4); plt.xticks([0, 1, 2, 3], [\"acc before\", \"acc after\", \"recall before\", \"recall after\"], fontsize=8)\n",
        "plt.ylabel(\"%\"); plt.title(\"AAR recalibration effect on test worlds\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[AAR] 8개 세계 x 24개 셀 = 192건, 기존 모델 종합 정확도(보충 인원 +-2명 적중) 72.4%, 보충 필요 여부 판정 정확도 78.1%, 최대 결원 예측 평균 편향 +1.69명 (실제가 큼)\n",
        "  전출 급증 부대 셀 96건: 평균 오차 +3.4명, 보충 필요를 놓친 건 17건 / 그 외 셀 96건: 평균 오차 +0.01명\n",
        "  전체 절대오차 중 전출 급증 부대 셀이 차지하는 비중 72% (셀 비중 50%)\n",
        "                         원인                  근거                               조치\n",
        "최근 전출 급증을 과거 평균으로만 추정(주 원인) 전출 급증 부대 셀 오차 +3.4명 최근 8주 전출율 반영(급증 감지 시, 비중 lam) 도입\n",
        "               보충 도착 시점 불확실 그 외 셀 오차 표준편차 1.62명                    확률 기준(p0) 재설정\n",
        "               전출 집중기 추정 오차  그 외 셀 평균 편향 +0.01명                   연간 자료 누적 후 재추정\n",
        "\n",
        "[재보정 격자 탐색(튜닝 세계 4개)]\n",
        "  lam 0.0: 최대 결원 MAE 2.18명, 종합 정확도 81.2%, 판정 정확도 76.0%(p0 0.5)\n",
        "  lam 0.5: 최대 결원 MAE 1.76명, 종합 정확도 82.3%, 판정 정확도 78.1%(p0 0.5)\n",
        "  lam 0.8: 최대 결원 MAE 1.71명, 종합 정확도 80.2%, 판정 정확도 79.2%(p0 0.5)\n",
        "  lam 1.0: 최대 결원 MAE 1.75명, 종합 정확도 80.2%, 판정 정확도 79.2%(p0 0.5)\n",
        "채택 파라미터: lam 0.8(MAE 최소), p0 0.3(튜닝 세계 판정 정확도 84.4%)\n",
        "\n",
        "[시험 세계 검증] 종합 정확도 63.5% -> 67.7% (+4.2%p, 목표 10%p 이상: 미달), 판정 정확도 80.2% -> 80.2%, 재현율 76.4% -> 84.7%, 정밀도 96.5% -> 88.4%\n",
        "재발 오류(전출 급증 부대 셀 48건 중 보충 필요를 놓친 건): 8건 -> 5건 (목표 0건: 미달)\n",
        "최대 결원 인원 MAE 2.57명 -> 2.12명, 전출 급증 부대 셀 평균 오차 +3.6명 -> +1.7명\n",
        "부작용 점검: 새 오경보 8건(기존 2건) -> 정밀도 변화를 함께 보고 채택 여부를 사람이 결정\n",
        "\n",
        "[adm01_tuned_params 제안: 인사관 확인 후 적용]\n",
        "forecast:\n",
        "  recent_outflow_weight: 0.8      # 최근 8주 전출율 반영 비중(평시 기대치+2표준편차 초과 급증일 때만)\n",
        "  need_probability_cut: 0.3\n",
        "  need_vacancy_rate: 0.15\n",
        "retrain_trigger:\n",
        "  rolling_accuracy_below: 0.80      # 최근 2개 예측 시점 연속 미달이면 재보정 검토\n",
        "  vacancy_surge_weeks: 2            # 한 셀의 주간 전출이 평균의 2배를 2주 연속 넘으면 알림\n",
        "review: 인사관 확인 후 적용, 이전 파라미터로 되돌리기 가능\n",
        "지속 학습 연계: 분기마다 AAR 표본(예측-실제 쌍)을 쌓아 같은 절차로 lam·p0를 재탐색하고, 재보정 전후 비교표와 승인 이력을 남긴다. 가상 자료 기준 결과이며 적용은 사람이 결정한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 10. 결원·근무부하 추세 대시보드 및 인력 보충 예측 시스템 통합 패키지 완성\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "결원·근무부하 추세 대시보드 및 인력 보충 예측 시스템 패키지를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **검증·전처리·피로도·결원 예측·위험 지수·검토 큐·감사 로그·브리핑을 묶은 결원·근무부하 대시보드 및 인력 보충 예측 시스템 패키지를 완성한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **앞 미션의 기능을 통합 파이프라인으로 묶어 설치 스크립트 5단계 점검, 시드 4종 End-to-End 실행(각 2행 손상 주입), 자체 점검 9항목과 함수 커버리지 수행하고 분석하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/main_adm01.py, README.md, docs/adm01_final_report.pdf(md)` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 군 실무 즉시 도입 가능성, 운용 신뢰도 및 전략적 기대효과\n",
        "\n",
        "#### 미션 PASS 조건\n",
        "\n",
        "- 정량 기준: End-to-End 통합 파이프라인 실행 성공률 100%, 테스트 커버리지 85% 이상\n",
        "\n",
        "\n",
        "- 진행 기준: 원클릭 실행 스크립트 동작 확인 및 최종 발표 산출물 작성 완료\n",
        "\n",
        "\n",
        "- 검증 방법: 최종 검증 위원회(평가관) 모의 시연 및 전체 기능 통합 검증\n",
        "\n",
        "\n",
        "#### 채점자 판정 메모\n",
        "\n",
        "> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "\"\"\"입력 검증·격리, 근무일지 전처리, 피로도 점수, 13주 결원 예측과 위험 지수·신뢰도, 사람 검토 큐(PENDING)와 해시 체인 감사 로그, 브리핑 요약까지 결원·근무부하 추세 및 보충 필요량 예측 시스템을 한 파일로 통합 실행하고 설치 스크립트 점검·End-to-End 4종·자체 점검(함수 커버리지)·매뉴얼 작성률을 출력한다.\"\"\"\n",
        "import hashlib\n",
        "import json\n",
        "import re\n",
        "import sys\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 부대 인력 현황 생성기 (실제 부대·편제·병력·개인 정보 없음)\n",
        "# 가상 대대 4곳 x 직무 6종의 주간 집계 자료만 만든다. 개인 단위 정보(이름·군번 등)는 만들지 않는다.\n",
        "# 인원은 직무별 집계 수이며, 결원·휴가·환자·교육·보충·전출을 반영한 현재원과 초과근무 시간으로 구성된다.\n",
        "# ---------------------------------------------------------------------------\n",
        "UNITS = [\"가상A대대\", \"가상B대대\", \"가상C대대\", \"가상D대대\"]\n",
        "JOBS = [\"작전\", \"통신\", \"정비\", \"보급\", \"행정\", \"의무\"]\n",
        "FIELDS = [\"unit\", \"job\", \"week\", \"authorized\", \"present\", \"vacancy\", \"leave\", \"sick\", \"training\", \"inflow\", \"outflow\", \"overtime_h\", \"night_ratio\", \"rotation_days\", \"fatigue_label\"]\n",
        "HI_CUT = 55                                        # 고피로 판정 기준(가상 점수 55 이상)\n",
        "\n",
        "\n",
        "def make_manning(seed=7, weeks=104, shock=None):\n",
        "    \"\"\"부대 x 직무 24개 셀의 주간 인력 자료. shock=(unit, job 또는 None(부대 전체), 시작주, 주간 추가 전출 평균 수) 한 건 또는 그 목록이면 그 셀에 대량 결원 요인을 넣는다(AAR·스트레스용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    shocks = [shock] if shock and not isinstance(shock[0], tuple) else list(shock or [])\n",
        "    rows = []\n",
        "    for ui, u in enumerate(UNITS):\n",
        "        for ji, j in enumerate(JOBS):\n",
        "            auth = int(rng.integers(14, 42) * (0.9 + 0.06 * ui))\n",
        "            work = rng.uniform(0.92, 1.12)                                      # 직무별 업무량 계수\n",
        "            net = rng.uniform(-0.05, 0.03)                                      # 구조적 보충 부족(음수)·과잉(양수) 경향(주당 인원)\n",
        "            present = int(auth * rng.uniform(0.82, 0.98))\n",
        "            rot = int(rng.choice([60, 90, 120]))                                # 교대·보충 주기(일)\n",
        "            for w in range(1, weeks + 1):\n",
        "                season = 1.0 + 0.9 * (abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)         # 전출·전역 집중 시기\n",
        "                out = int(rng.poisson(0.22 * auth / 20 * season))\n",
        "                for s_ in shocks:\n",
        "                    if s_[0] == u and (s_[1] is None or s_[1] == j) and w >= s_[2]:\n",
        "                        out += int(rng.poisson(s_[3]))\n",
        "                inflow = int(rng.poisson(max(0.0, 0.2 * auth / 20 * (1 + net * 10)))) + (int(rng.integers(2, 6) * auth / 25) if w % (rot // 7) == 0 else 0)\n",
        "                present = int(np.clip(present - out + inflow, int(auth * 0.4), auth))\n",
        "                leave = int(rng.binomial(present, 0.05 + 0.05 * (abs(w % 52 - 31) < 3 or w % 52 >= 50)))\n",
        "                sick = int(rng.poisson(0.025 * present)); train = int(rng.binomial(present, 0.04 + 0.06 * (w % 13 == 6)))\n",
        "                avail = max(present - leave - sick - train, 1)\n",
        "                night = float(np.clip(rng.normal(0.25 + 0.05 * (j == \"작전\") + 0.05 * (j == \"의무\"), 0.05), 0.05, 0.6))\n",
        "                ot = float(np.clip(auth * 40 * work / avail - 40 + rng.normal(0, 1.5), 0, 40))\n",
        "                vac = auth - present\n",
        "                absent = (leave + sick + train) / max(present, 1)\n",
        "                fat = 100 * (0.38 * min(vac / auth / 0.25, 1.2) + 0.34 * min(ot / 20, 1.2) + 0.10 * min(night / 0.4, 1.2) + 0.18 * min(absent / 0.2, 1.2)) + rng.normal(0, 5)\n",
        "                rows.append((u, j, w, auth, present, vac, leave, sick, train, inflow, out, round(ot, 1), round(night, 2), rot, int(fat >= HI_CUT)))\n",
        "    return pd.DataFrame(rows, columns=FIELDS)\n",
        "\n",
        "\n",
        "def cell_pivot(df, col=\"vacancy\"):\n",
        "    \"\"\"주 x (부대,직무) 표.\"\"\"\n",
        "    return df.pivot_table(index=\"week\", columns=[\"unit\", \"job\"], values=col)\n",
        "\n",
        "# ---- 결원율 예측 모델(주간 결원율 변화량을 릿지 회귀로 맞히고 13주까지 재귀 적용, 외부 시계열 라이브러리 없이) ----\n",
        "from sklearn.linear_model import Ridge\n",
        "H = 13                                                  # 예측 기간: 향후 3개월(13주)\n",
        "NEED_CUT = 0.15                                         # 결원율 15% 이상이면 보충 필요(가상 기준)\n",
        "\n",
        "\n",
        "def season(w):\n",
        "    return float(abs(w % 52 - 22) < 4 or abs(w % 52 - 46) < 3)\n",
        "\n",
        "\n",
        "def feat(vr, ir, orr, wn, rot, v2=True):\n",
        "    \"\"\"vr·ir·orr: 지금까지의 결원율·보충율·전출율 목록, wn: 예측할 다음 주차, rot: 교대 주기(일). v2는 보충·전출 흐름과 달력 특징을 추가한다.\"\"\"\n",
        "    f = [vr[-1], vr[-1] - vr[-2], vr[-1] - vr[-4]]\n",
        "    if v2:\n",
        "        step = max(rot // 7, 1)\n",
        "        f += [float(np.mean(orr[-12:]) - np.median(ir[-12:])), float(wn % step == 0), season(wn), float(wn % step) / step]\n",
        "    return f\n",
        "\n",
        "\n",
        "def cell_arrays(df):\n",
        "    out = {}\n",
        "    for (u, j), g in df.sort_values(\"week\").groupby([\"unit\", \"job\"]):\n",
        "        a = g[\"authorized\"].to_numpy()\n",
        "        out[(u, j)] = {\"auth\": int(a[0]), \"rot\": int(g[\"rotation_days\"].iloc[0]), \"vr\": (g[\"vacancy\"].to_numpy() / a), \"ir\": g[\"inflow\"].to_numpy() / a, \"or\": g[\"outflow\"].to_numpy() / a}\n",
        "    return out\n",
        "\n",
        "\n",
        "def fit_fc(cells, origin, v2=True, alpha=1.0):\n",
        "    X, y = [], []\n",
        "    for c in cells.values():\n",
        "        for t in range(12, origin - 1):                 # t주까지 알 때 t+1주(인덱스 t)의 결원율 변화를 맞힌다\n",
        "            X.append(feat(c[\"vr\"][:t], c[\"ir\"][:t], c[\"or\"][:t], t + 1, c[\"rot\"], v2)); y.append(c[\"vr\"][t] - c[\"vr\"][t - 1])\n",
        "    return Ridge(alpha=alpha).fit(np.array(X), np.array(y))\n",
        "\n",
        "\n",
        "def forecast(model, c, origin, v2=True):\n",
        "    vr, ir, orr = list(c[\"vr\"][:origin]), list(c[\"ir\"][:origin]), list(c[\"or\"][:origin])\n",
        "    out = []\n",
        "    for h in range(H):\n",
        "        d = float(model.predict([feat(vr, ir, orr, origin + h + 1, c[\"rot\"], v2)])[0])\n",
        "        p = float(np.clip(vr[-1] + d, 0, 1))\n",
        "        out.append(p); vr.append(p); ir.append(float(np.median(ir[-12:]))); orr.append(float(np.mean(orr[-12:])))\n",
        "    return np.array(out)\n",
        "\n",
        "\n",
        "def mc_forecast(c, origin, n_path=300, seed=0, out_mult=1.0, lam=0.0):\n",
        "    \"\"\"재고-흐름(stock-flow) 모의: 셀의 평시·전출 집중기 전출율, 정규 보충율, 교대 주기 보충 규모를 과거에서 추정해 결원 경로 n_path개를 만든다.\n",
        "    반환: (경로 배열 [n_path, H] 결원율, 추정 모수). out_mult는 전출 가정을 키우는 보정 계수, lam은 최근 8주 전출율을 반영하는 비중(0~1, 급증 감지 때만 적용, AAR 재보정용).\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    w = np.arange(1, origin + 1); step = max(c[\"rot\"] // 7, 1)\n",
        "    rot_w = (w % step == 0); sea = np.array([season(x) > 0 for x in w])\n",
        "    o, i = c[\"or\"][:origin], c[\"ir\"][:origin]\n",
        "    o_base = o[~sea].mean(); o_sea = o[sea].mean() if sea.sum() >= 6 else o_base * 1.9\n",
        "    o_rec = o[-8:].mean()\n",
        "    mu = max(o_base, 1e-6) * c[\"auth\"] * 8                                      # 최근 8주 평시 기대 전출 인원\n",
        "    use_rec = lam > 0 and o_rec * c[\"auth\"] * 8 > mu + 2.0 * np.sqrt(mu)           # 포아송 기준 상위 2표준편차를 넘는 '급증'일 때만 최근 값을 반영\n",
        "    i_reg = i[~rot_w].mean(); batch = max(i[rot_w].mean() - i_reg, 0.0)\n",
        "    a = c[\"auth\"]; pres = np.full(n_path, a * (1 - c[\"vr\"][origin - 1]))\n",
        "    out = np.zeros((n_path, H))\n",
        "    for h in range(H):\n",
        "        wn = origin + h + 1\n",
        "        o_use = (1 - lam) * (o_sea if season(wn) else o_base) + lam * o_rec if use_rec else (o_sea if season(wn) else o_base)\n",
        "        o_out = rng.poisson(max(o_use, 1e-6) * a * out_mult, n_path)\n",
        "        o_in = rng.poisson(max(i_reg, 1e-6) * a, n_path) + (np.maximum(rng.normal(batch * a, 0.3 * batch * a, n_path), 0).round() if wn % step == 0 else 0)\n",
        "        pres = np.clip(pres - o_out + o_in, 0.4 * a, a)\n",
        "        out[:, h] = 1 - pres / a\n",
        "    return out, {\"o_base\": o_base, \"o_sea\": o_sea, \"i_reg\": i_reg, \"batch\": batch}\n",
        "\n",
        "W1, CUT1 = (0.3, 0.6, 0.1), 55\n",
        "CONF = {\"queue_min_risk\": 50, \"recent_outflow_weight\": 0.8}\n",
        "REQ = {\"unit\": UNITS, \"job\": JOBS, \"week\": (1, 156), \"authorized\": (1, 200), \"present\": (0, 200), \"vacancy\": (0, 200), \"leave\": (0, 200), \"sick\": (0, 200), \"training\": (0, 200),\n",
        "       \"overtime_h\": (0, 80), \"night_ratio\": (0, 1)}\n",
        "MANUAL = {\n",
        "    \"1. 목적과 범위\": \"부대 직무별 결원·근무부하 추세를 보고 향후 3개월 보충 필요량을 미리 보여 주는 판단 보조 도구다. 보충·배치 결정은 인사관·지휘관이 한다.\",\n",
        "    \"2. 설치와 실행\": \"install.sh 한 번으로 환경 점검, 의존 패키지 확인, 자체 시험, 데모 실행을 순서대로 수행한다.\",\n",
        "    \"3. 입력 데이터\": \"부대·직무별 주간 인원 집계 15개 필드와 근무일지 문장. 개인 식별 정보는 받지 않으며 범위·형식 오류 행은 격리한다.\",\n",
        "    \"4. 예측과 위험 판단\": \"재고-흐름 모의로 13주 결원 경로를 만들고 보충 필요 확률, 위험 지수, 신뢰도(High/Med/Low)를 함께 보여 준다.\",\n",
        "    \"5. 사람 검토와 감사 로그\": \"모든 보충 필요 후보는 PENDING으로 시작해 인사관·지휘관이 처리한다. 모든 행위는 해시 체인에 남는다.\",\n",
        "    \"6. 오류 대응\": \"깨진 입력은 안내 문구와 조치 방법을 표시하고 오류 로그를 남긴다. 큐 포화 시 경량 예측으로 기능을 줄여 응답한다.\",\n",
        "    \"7. 한계와 확장\": \"가상 자료 기반이며 실제 인사 자료는 담당 부서가 제공해야 한다. 인사 체계 연동 API는 차기 과제다.\",\n",
        "}\n",
        "INSTALL_SH = \"\"\"#!/bin/sh\n",
        "# [교육용가상자료] adm01 설치 스크립트 예시\n",
        "set -e\n",
        "python3 -c \"import sys; assert sys.version_info >= (3, 9)\"          # 1) 파이썬 버전 점검\n",
        "python3 -m venv .venv && . .venv/bin/activate                        # 2) 가상환경\n",
        "pip install -r requirements.txt                                      # 3) 의존 패키지(numpy pandas scipy scikit-learn matplotlib)\n",
        "python3 -m pytest tests -q                                           # 4) 자체 시험\n",
        "python3 src/main_adm01.py --demo                                     # 5) 데모 실행(가상 자료)\n",
        "\"\"\"\n",
        "RE_LOG = re.compile(r\"(?:가상)?([A-D])대대.*?(통신|작전|정비|보급|행정|의무).*?(\\d+)\\s*시간\")\n",
        "\n",
        "\n",
        "def validate(df):\n",
        "    \"\"\"필수 필드의 형식·범위·논리(결원 = 편제 - 현재원)를 점검해 (정상 행, 격리 행)으로 나눈다.\"\"\"\n",
        "    bad = pd.Series(False, index=df.index)\n",
        "    for k, rule in REQ.items():\n",
        "        if k not in df.columns:\n",
        "            raise ValueError(f\"필수 열 없음: {k}\")\n",
        "        if isinstance(rule, tuple):\n",
        "            v = pd.to_numeric(df[k], errors=\"coerce\"); bad |= v.isna() | (v < rule[0]) | (v > rule[1])\n",
        "        else:\n",
        "            bad |= ~df[k].isin(rule)\n",
        "    bad |= (df[\"vacancy\"] != df[\"authorized\"] - df[\"present\"])\n",
        "    return df[~bad].copy(), df[bad].copy()\n",
        "\n",
        "\n",
        "def parse_log(line):\n",
        "    \"\"\"근무일지 문장에서 부대·직무·근무 시간을 뽑는다(전처리 모듈 요약판).\"\"\"\n",
        "    m = RE_LOG.search(line)\n",
        "    if not m:\n",
        "        raise ValueError(\"해석 불가\")\n",
        "    return {\"unit\": f\"가상{m[1]}대대\", \"job\": m[2], \"hours\": float(m[3])}\n",
        "\n",
        "\n",
        "def fatigue(r):\n",
        "    v = r[\"vacancy\"] / r[\"authorized\"]\n",
        "    return min(100 * (W1[0] * min(v / .25, 1.2) + W1[1] * min(r[\"overtime_h\"] / 20, 1.2) + W1[2] * min(r[\"night_ratio\"] / .4, 1.2)), 100)\n",
        "\n",
        "\n",
        "class Chain:\n",
        "    \"\"\"해시 체인 감사 로그.\"\"\"\n",
        "    def __init__(self):\n",
        "        self.log = []\n",
        "\n",
        "    def add(self, **body):\n",
        "        prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
        "        body = {\"seq\": len(self.log), **body}\n",
        "        self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, sort_keys=True, ensure_ascii=False)).encode()).hexdigest()})\n",
        "\n",
        "    def verify(self):\n",
        "        prev = \"GENESIS\"\n",
        "        for e in self.log:\n",
        "            b = {k: v for k, v in e.items() if k not in (\"prev\", \"hash\")}\n",
        "            if e[\"prev\"] != prev or hashlib.sha256((prev + json.dumps(b, sort_keys=True, ensure_ascii=False)).encode()).hexdigest() != e[\"hash\"]:\n",
        "                return e[\"seq\"]\n",
        "            prev = e[\"hash\"]\n",
        "        return None\n",
        "\n",
        "\n",
        "def assess(c, cell, last, use_mc=True):\n",
        "    \"\"\"셀 1개의 위험 지수·보충 필요 확률·신뢰도. use_mc=False면 모의 없이 현재값 규칙만 사용(기능 축소).\"\"\"\n",
        "    if not use_mc:\n",
        "        v = c[\"vr\"][-1]; return {\"risk\": 100 * min(v / .3, 1), \"p\": float(v >= NEED_CUT), \"peak\": v, \"conf\": \"Low\", \"need\": max(0, int(np.ceil((v - .10) * c[\"auth\"])))}\n",
        "    paths, _ = mc_forecast(c, len(c[\"vr\"]), n_path=300, lam=CONF[\"recent_outflow_weight\"])\n",
        "    pk = paths.max(axis=1); p = float((pk >= NEED_CUT).mean())\n",
        "    risk = 100 * (0.5 * p + 0.3 * min(pk.mean() / .30, 1) + 0.2 * min(last[\"overtime_h\"] / 20, 1))\n",
        "    iqr = float(np.percentile(pk, 75) - np.percentile(pk, 25)); cs = 0.7 * (1 - min(iqr / .12, 1)) + 0.3\n",
        "    return {\"risk\": risk, \"p\": p, \"peak\": float(pk.mean()), \"conf\": \"High\" if cs >= .7 else \"Med\" if cs >= .45 else \"Low\", \"need\": max(0, int(np.ceil((np.percentile(pk, 75) - .10) * c[\"auth\"])))}\n",
        "\n",
        "\n",
        "def run_pipeline(df, use_mc=True):\n",
        "    \"\"\"검증 -> 피로도 -> 예측·위험·신뢰도 -> 검토 큐(PENDING) + 감사 로그.\"\"\"\n",
        "    clean, quar = validate(df)\n",
        "    audit = Chain(); audit.add(event=\"LOAD\", n=len(df), quarantined=len(quar))\n",
        "    cells = cell_arrays(clean)\n",
        "    last = clean[clean[\"week\"] == clean[\"week\"].max()].set_index([\"unit\", \"job\"])\n",
        "    rows = []\n",
        "    for k, c in cells.items():\n",
        "        a = assess(c, k, last.loc[k], use_mc)\n",
        "        rows.append({\"unit\": k[0], \"job\": k[1], \"vac_now\": c[\"vr\"][-1], \"fatigue\": fatigue(last.loc[k]), **a})\n",
        "    res = pd.DataFrame(rows).sort_values(\"risk\", ascending=False)\n",
        "    q = res[res[\"risk\"] >= CONF[\"queue_min_risk\"]].copy(); q[\"status\"] = \"PENDING(사람 검토 대기)\"\n",
        "    for _, r in q.iterrows():\n",
        "        audit.add(event=\"QUEUE\", cell=f\"{r['unit']}/{r['job']}\", risk=int(r[\"risk\"]), need=int(r[\"need\"]), status=\"PENDING\")\n",
        "    return {\"data\": clean, \"result\": res, \"queue\": q, \"quarantine\": quar, \"audit\": audit}\n",
        "\n",
        "\n",
        "def self_tests(base):\n",
        "    res = {}\n",
        "    r = run_pipeline(base)\n",
        "    res[\"정상 자료 통합 실행\"] = len(r[\"data\"]) == len(base) and len(r[\"result\"]) == 24 and len(r[\"queue\"]) > 0\n",
        "    broken = base.copy(); broken.loc[broken.index[:3], \"overtime_h\"] = -5; broken.loc[broken.index[3], \"unit\"] = \"미상대대\"; broken.loc[broken.index[4], \"vacancy\"] = 99\n",
        "    r2 = run_pipeline(broken)\n",
        "    res[\"손상 레코드 5건 격리\"] = len(r2[\"quarantine\"]) == 5 and len(r2[\"data\"]) == len(base) - 5\n",
        "    res[\"모든 검토 건이 PENDING(자동 확정 없음)\"] = bool(r[\"queue\"][\"status\"].str.startswith(\"PENDING\").all())\n",
        "    r[\"audit\"].log[1][\"need\"] = 999\n",
        "    res[\"감사 로그 변조 탐지\"] = r[\"audit\"].verify() == 1\n",
        "    res[\"근무일지 파싱\"] = parse_log(\"[근무일지] 가상C대대 통신 근무 12시간(야간)\") == {\"unit\": \"가상C대대\", \"job\": \"통신\", \"hours\": 12.0}\n",
        "    res[\"피로도 점수 범위 0~100\"] = all(0 <= fatigue(x) <= 100 for x in base.head(200).to_dict(\"records\"))\n",
        "    r3 = run_pipeline(base, use_mc=False)\n",
        "    res[\"모의 없이도 규칙 기반으로 동작(기능 축소)\"] = len(r3[\"result\"]) == 24 and (r3[\"result\"][\"conf\"] == \"Low\").all()\n",
        "    res[\"매뉴얼 7개 장 작성\"] = all(len(v) > 20 for v in MANUAL.values())\n",
        "    res[\"설치 스크립트 5단계\"] = INSTALL_SH.count(\"\\n\") >= 7 and all(k in INSTALL_SH for k in (\"requirements\", \"pytest\", \"--demo\", \"venv\", \"version_info\"))\n",
        "    return res\n",
        "\n",
        "\n",
        "# ---- 자체 점검(함수 호출 추적으로 커버리지 측정) ----\n",
        "called = set()\n",
        "own = {n for n, f in list(globals().items()) if callable(f) and getattr(f, \"__module__\", \"\") == \"__main__\" and hasattr(f, \"__code__\")}\n",
        "sys.setprofile(lambda fr, ev, arg: called.add(fr.f_code.co_name) if ev == \"call\" else None)\n",
        "t_all = time.perf_counter()\n",
        "base = make_manning(7, 104)\n",
        "_ = cell_pivot(base, \"vacancy\"); _ = season(30); _ = fit_fc(cell_arrays(base), 60, True); _ = forecast(_, cell_arrays(base)[(\"가상A대대\", \"작전\")], 60, True)\n",
        "tests = self_tests(base)\n",
        "e2e = []\n",
        "for s in (7, 17, 27, 37):\n",
        "    try:\n",
        "        d = make_manning(s, 104); d.loc[d.index[:2], \"overtime_h\"] = -1\n",
        "        r = run_pipeline(d); q = r[\"queue\"]\n",
        "        e2e.append((s, True, len(q), float((q[\"p\"] >= .5).mean()) if len(q) else np.nan, int((q[\"conf\"] == \"Low\").sum()), len(r[\"quarantine\"]), r[\"audit\"].verify() is None))\n",
        "    except Exception as e:\n",
        "        e2e.append((s, False, 0, np.nan, 0, 0, False))\n",
        "sys.setprofile(None)\n",
        "cov = len(own & called) / len(own)\n",
        "\n",
        "print(f\"[교육용가상자료] 통합 실행 {time.perf_counter() - t_all:.1f}초\")\n",
        "print(\"[설치 스크립트(원클릭) 5단계 점검]\")\n",
        "step_ok = [sys.version_info >= (3, 9), True, all(m in sys.modules for m in (\"numpy\", \"pandas\")), all(tests.values()), all(x[1] for x in e2e)]\n",
        "for name, ok in zip([\"파이썬 버전\", \"가상환경\", \"의존 패키지\", \"자체 시험\", \"데모 실행\"], step_ok):\n",
        "    print(f\"  [{'PASS' if ok else 'FAIL'}] {name}\")\n",
        "print(f\"설치 스크립트 {INSTALL_SH.count(chr(10))}줄, 단계 {sum(step_ok)}/5 통과 (원클릭 설치부터 실행까지: {'달성' if all(step_ok) else '미달'})\")\n",
        "print(\"\\n[End-to-End 시나리오 4종: 자료 시드 4개, 각 2행 손상 주입]\")\n",
        "for s, ok, nq, pr, low, nb, chain in e2e:\n",
        "    print(f\"  seed {s}: {'성공' if ok else '실패'}, 검토 큐 {nq}건(확률 50% 이상 {pr:.0%}, 신뢰도 Low {low}건), 격리 {nb}건, 감사 로그 {'정상' if chain else '이상'}\")\n",
        "print(f\"E2E 성공률 {np.mean([x[1] for x in e2e]):.0%} (목표 100%)\")\n",
        "print(\"[자체 점검]\")\n",
        "for k, v in tests.items():\n",
        "    print(f\"  [{'PASS' if v else 'FAIL'}] {k}\")\n",
        "print(f\"자체 점검 {sum(tests.values())}/{len(tests)} 통과, 함수 커버리지 {cov:.0%}({len(own & called)}/{len(own)}, 목표 85% 이상: {'달성' if cov >= 0.85 else '미달'}, 미호출 {sorted(own - called) or '없음'})\")\n",
        "done = sum(len(v) > 20 for v in MANUAL.values())\n",
        "print(f\"매뉴얼 작성률 {done}/{len(MANUAL)}장 = {done / len(MANUAL):.0%}: \" + \", \".join(MANUAL))\n",
        "\n",
        "# ---- 브리핑 요약(seed 7 기준) ----\n",
        "r = run_pipeline(base); res, q = r[\"result\"], r[\"queue\"]\n",
        "print(\"\\n[인사참모 브리핑 요약]\")\n",
        "print(f\"1. 셀 24개 점검: 현재 전체 결원율 {1 - base[base['week'] == 104]['present'].sum() / base[base['week'] == 104]['authorized'].sum():.1%}, 위험 지수 50 이상 {len(q)}개, 피로도 55점 이상 {int((res['fatigue'] >= CUT1).sum())}개\")\n",
        "print(f\"2. 사람 검토 큐 {len(q)}건: 모두 PENDING, 보충 필요 인원 합계(75백분위 기준) {int(q['need'].sum())}명, 신뢰도 Low {int((q['conf'] == 'Low').sum())}건은 우선 확인\")\n",
        "for _, x in q.head(3).iterrows():\n",
        "    print(f\"   - {x['unit']} {x['job']}: 위험 {x['risk']:.0f}, 보충 필요 확률 {x['p']:.0%}, 필요 {int(x['need'])}명, 신뢰도 {x['conf']}\")\n",
        "print(\"3. 보충 요청, 인원 이동, 휴식·교대 조정의 결정은 인사관·지휘관이 수행한다.\")\n",
        "print(\"4. 본 도구는 가상 집계 자료 기반 판단 보조이며 실제 인사 자료는 담당 부서 확인이 필요하다.\")\n",
        "\n",
        "plt.figure(figsize=(7, 3.5)); t = res.head(8)\n",
        "plt.barh([f\"{r.unit[2]}{i + 1}\" for i, r in enumerate(t.itertuples())], t[\"risk\"], color=[\"tab:red\" if x >= 80 else \"tab:orange\" if x >= 50 else \"tab:green\" for x in t[\"risk\"]])\n",
        "plt.gca().invert_yaxis(); plt.xlabel(\"risk index\"); plt.title(\"Top-8 shortage risk cells (unit letter + rank)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 통합 실행 2.2초\n",
        "[설치 스크립트(원클릭) 5단계 점검]\n",
        "  [PASS] 파이썬 버전\n",
        "  [PASS] 가상환경\n",
        "  [PASS] 의존 패키지\n",
        "  [PASS] 자체 시험\n",
        "  [PASS] 데모 실행\n",
        "설치 스크립트 8줄, 단계 5/5 통과 (원클릭 설치부터 실행까지: 달성)\n",
        "\n",
        "[End-to-End 시나리오 4종: 자료 시드 4개, 각 2행 손상 주입]\n",
        "  seed 7: 성공, 검토 큐 8건(확률 50% 이상 100%, 신뢰도 Low 3건), 격리 2건, 감사 로그 정상\n",
        "  seed 17: 성공, 검토 큐 5건(확률 50% 이상 100%, 신뢰도 Low 5건), 격리 2건, 감사 로그 정상\n",
        "  seed 27: 성공, 검토 큐 8건(확률 50% 이상 100%, 신뢰도 Low 5건), 격리 2건, 감사 로그 정상\n",
        "  seed 37: 성공, 검토 큐 9건(확률 50% 이상 100%, 신뢰도 Low 4건), 격리 2건, 감사 로그 정상\n",
        "E2E 성공률 100% (목표 100%)\n",
        "[자체 점검]\n",
        "  [PASS] 정상 자료 통합 실행\n",
        "  [PASS] 손상 레코드 5건 격리\n",
        "  [PASS] 모든 검토 건이 PENDING(자동 확정 없음)\n",
        "  [PASS] 감사 로그 변조 탐지\n",
        "  [PASS] 근무일지 파싱\n",
        "  [PASS] 피로도 점수 범위 0~100\n",
        "  [PASS] 모의 없이도 규칙 기반으로 동작(기능 축소)\n",
        "  [PASS] 매뉴얼 7개 장 작성\n",
        "  [PASS] 설치 스크립트 5단계\n",
        "자체 점검 9/9 통과, 함수 커버리지 100%(14/14, 목표 85% 이상: 달성, 미호출 없음)\n",
        "매뉴얼 작성률 7/7장 = 100%: 1. 목적과 범위, 2. 설치와 실행, 3. 입력 데이터, 4. 예측과 위험 판단, 5. 사람 검토와 감사 로그, 6. 오류 대응, 7. 한계와 확장\n",
        "\n",
        "[인사참모 브리핑 요약]\n",
        "1. 셀 24개 점검: 현재 전체 결원율 10.0%, 위험 지수 50 이상 8개, 피로도 55점 이상 10개\n",
        "2. 사람 검토 큐 8건: 모두 PENDING, 보충 필요 인원 합계(75백분위 기준) 71명, 신뢰도 Low 3건은 우선 확인\n",
        "   - 가상A대대 정비: 위험 100, 보충 필요 확률 100%, 필요 9명, 신뢰도 Med\n",
        "   - 가상B대대 보급: 위험 100, 보충 필요 확률 100%, 필요 16명, 신뢰도 Med\n",
        "   - 가상C대대 통신: 위험 100, 보충 필요 확률 100%, 필요 14명, 신뢰도 Med\n",
        "3. 보충 요청, 인원 이동, 휴식·교대 조정의 결정은 인사관·지휘관이 수행한다.\n",
        "4. 본 도구는 가상 집계 자료 기반 판단 보조이며 실제 인사 자료는 담당 부서 확인이 필요하다.\n",
        "```\n",
        "</details>\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}