{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# 🎯 [DAT10] 데이터 마이닝 기반 공급망 재난 식별 SaaS\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/DAT10/DAT10_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",
        "전시에는 통신망이 끊긴 채 보급소가 입고와 불출을 각자 기록하고, 복구 후 여러 기록을 합치는 과정에서 중복·누락·충돌·위변조가 생기며, 이를 놓치면 재고 판단과 업체 공급 이상 징후 식별이 늦어진다. 본 프로젝트는 방산 공급망 회복력 연구(미국 방위산업기반 정책연구)의 문제의식을 교육용 가상 자료로 재구성해, 가상 보급소 5곳이 오프라인으로 재고를 기록하고 동기화하는 SaaS를 만들고 동기화된 자료에서 공급망 재난 징후(업체 공급 감소, 재고 소진 임박)를 식별하는 과제이다. 요구사항·스키마, 오프라인 입력 전처리, P2P/USB 교환, CRDT 동기화와 충돌 탐지, 무결성 검증, 대시보드, 군수관 HITL 병합, 7일 단절 후 일괄 동기화 스트레스, 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. 전술 통신 두절 환경에서의 야전 군수 재고 관리 요구사항 분석 및 로컬 DB 스키마 설계\n",
        "**소요시간**: 약 1.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "통신 두절 환경에서의 야전 재고 관리 요구사항을 분석하고 로컬 DB 스키마를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **통신이 끊긴 환경에서 보급소가 입고·불출·조정을 기록하는 오프라인 재고 관리의 요구사항 12개를 정리하고, 이를 만족하는 트랜잭션 스키마(필드 12개)와 로컬 DB 구조를 설계한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **가상 보급소 5곳과 품목 12종의 트랜잭션(ID·보급소·품목·구분·수량·시각·상황·업체·논리 시계·동기화 상태·체크섬·비고) 필드를 자료형·단위·허용 범위와 함께 정의하고, ID는 노드 접두어와 일련번호로 만들어 중복 입력을 막는 멱등 키로 쓴다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `data/schema/dat10_input_schema.json 및 docs/dat10_data_spec.md` 형태의 결과물을 완성한다.\n",
        "\n",
        "### 🔍 평가 및 검증 기준 (PASS 조건)\n",
        "#### 미션 판정\n",
        "\n",
        "**평가 구분:** 필수\n",
        "\n",
        "**핵심 평가영역:** 데이터·자동화·업무플랫폼 실무 및 공급망 재난 식별 데이터마이닝 SaaS(방산 공급망 회복력 연구) — 미국 방위산업기반(Defense Industrial Base) 정책연구 전장 환경과의 군사적 부합성\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": [
        "\"\"\"통신 두절 환경의 야전 보급소 재고·공급망 자료 요구사항 12개를 정의하고, 로컬 DB(SQLite 메모리 DB) 스키마와 트랜잭션 레코드 12개 필드의 검증 규칙을 만들어 정상 표본 검증 오류 0건·스키마 정의 완성도 100%·제약 조건(중복·음수·논리) 시험을 확인하며 오류 주입 검출력을 점검한다.\"\"\"\n",
        "import hashlib\n",
        "import json\n",
        "import re\n",
        "import sqlite3\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "# 1) 레코드 스키마(필드 12개): 자료형·허용값/범위·단위·설명을 모두 정의\n",
        "SCHEMA = {\n",
        "    \"id\": {\"type\": \"string\", \"pattern\": r\"^보급소[A-E]-\\d{5}$\", \"unit\": \"-\", \"desc\": \"트랜잭션 ID(노드-일련번호, 중복 입력 방지 키)\"},\n",
        "    \"node\": {\"type\": \"string\", \"enum\": NODES, \"unit\": \"-\", \"desc\": \"기록한 가상 보급소\"},\n",
        "    \"item\": {\"type\": \"string\", \"enum\": ITEM_IDS, \"unit\": \"-\", \"desc\": \"품목 번호(가상)\"},\n",
        "    \"op\": {\"type\": \"string\", \"enum\": [\"in\", \"out\", \"adjust\"], \"unit\": \"-\", \"desc\": \"입고·불출·조정\"},\n",
        "    \"qty\": {\"type\": \"integer\", \"min\": -50, \"max\": 5000, \"unit\": \"개\", \"desc\": \"수량(조정은 음수 허용)\"},\n",
        "    \"t\": {\"type\": \"number\", \"min\": 0, \"max\": 24 * 30, \"unit\": \"시간\", \"desc\": \"발생 시각(기준 시점부터 경과 시간)\"},\n",
        "    \"ctx\": {\"type\": \"string\", \"enum\": CTXS, \"unit\": \"-\", \"desc\": \"입력 상황(평시·야간·통신불량)\"},\n",
        "    \"supplier\": {\"type\": \"string\", \"enum\": SUPPLIERS, \"unit\": \"-\", \"desc\": \"가상 업체\"},\n",
        "    \"lamport\": {\"type\": \"integer\", \"min\": 0, \"max\": 10 ** 7, \"unit\": \"-\", \"desc\": \"논리 시계(노드별 증가)\"},\n",
        "    \"sync_state\": {\"type\": \"string\", \"enum\": [\"local\", \"sent\", \"acked\"], \"unit\": \"-\", \"desc\": \"동기화 상태\"},\n",
        "    \"checksum\": {\"type\": \"string\", \"pattern\": r\"^[0-9a-f]{8}$\", \"unit\": \"-\", \"desc\": \"위변조 점검용 해시 앞 8자리\"},\n",
        "    \"note\": {\"type\": \"string\", \"maxlen\": 40, \"unit\": \"-\", \"desc\": \"비고(자유 문장, 개인 정보 입력 금지)\"},\n",
        "}\n",
        "\n",
        "\n",
        "def seal(x, lamport):\n",
        "    body = f\"{x['id']}|{x['node']}|{x['item']}|{x['op']}|{x['qty']}|{x['t']}\"\n",
        "    return {**x, \"lamport\": lamport, \"sync_state\": \"local\", \"checksum\": hashlib.sha256(body.encode()).hexdigest()[:8], \"note\": \"\"}\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:\n",
        "            errs.append(f\"{k}: 필수 값 누락\"); continue\n",
        "        v = rec[k]\n",
        "        ok = {\"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:\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 \"pattern\" in r and not re.match(r[\"pattern\"], v):\n",
        "            errs.append(f\"{k}: 형식 불일치 '{v}'\")\n",
        "        if \"maxlen\" in r and len(v) > r[\"maxlen\"]:\n",
        "            errs.append(f\"{k}: 길이 {r['maxlen']} 초과\")\n",
        "    if not errs:\n",
        "        if rec[\"id\"].split(\"-\")[0] != rec[\"node\"]:\n",
        "            errs.append(\"id: 노드 접두어와 node 불일치\")\n",
        "        if rec[\"op\"] in (\"in\", \"out\") and rec[\"qty\"] <= 0:\n",
        "            errs.append(\"qty: 입고·불출 수량은 1 이상\")\n",
        "        if re.search(r\"\\d{6}-\\d{7}|군번|주민\", rec[\"note\"]):\n",
        "            errs.append(\"note: 개인 식별 정보 의심 문구\")\n",
        "    return errs\n",
        "\n",
        "\n",
        "txns = make_txns(7, 7)\n",
        "recs = [seal(x, i + 1) for i, x in enumerate(txns)]\n",
        "bad = sum(bool(check(r)) for r in recs)\n",
        "print(f\"[스키마] 필드 {len(SCHEMA)}개, type·unit·desc 모두 정의된 필드 {sum(all(k in v for k in ('type', 'unit', 'desc')) for v in SCHEMA.values())}개 (정의 완성도 100%)\")\n",
        "print(f\"[정상 자료 검증] 트랜잭션 {len(recs)}건 중 오류 {bad}건 (목표 오류 0건: {'달성' if bad == 0 else '미달'}), 개인 식별 필드 0개\")\n",
        "\n",
        "# 2) 오류 주입 시험\n",
        "r0 = recs[10]\n",
        "inj = {\"자료형 오류(qty 문자)\": dict(r0, qty=\"많이\"), \"허용값 오류(품목)\": dict(r0, item=\"I99\"), \"형식 오류(ID)\": dict(r0, id=\"A-1\"), \"범위 오류(시각)\": dict(r0, t=-3.0),\n",
        "       \"논리 오류(불출 0개)\": dict(r0, op=\"out\", qty=0), \"논리 오류(ID 접두어)\": dict(r0, node=\"보급소B\", id=\"보급소A-00011\"), \"필드 누락\": {k: v for k, v in r0.items() if k != \"checksum\"},\n",
        "       \"개인정보 의심 문구\": dict(r0, note=\"군번 입력\")}\n",
        "print(\"\\n[오류 주입 시험]\")\n",
        "caught = 0\n",
        "for n, r in inj.items():\n",
        "    e = check(r); caught += bool(e); print(f\"  {n:22s}-> {'검출' if e else '미검출'}: {e[0] if e else ''}\")\n",
        "print(f\"주입 오류 {len(inj)}종 중 {caught}종 검출\")\n",
        "\n",
        "# 3) 로컬 DB 스키마(SQLite 메모리 DB): 중복 키·음수 재고·논리 규칙을 DB가 직접 막는지 시험\n",
        "DDL = \"\"\"\n",
        "CREATE TABLE item (item TEXT PRIMARY KEY, cat TEXT NOT NULL, supplier TEXT NOT NULL, lead_days INTEGER CHECK (lead_days BETWEEN 1 AND 60), min_level INTEGER NOT NULL);\n",
        "CREATE TABLE txn (id TEXT PRIMARY KEY, node TEXT NOT NULL, item TEXT NOT NULL REFERENCES item(item), op TEXT NOT NULL CHECK (op IN ('in','out','adjust')),\n",
        "                  qty INTEGER NOT NULL, t REAL NOT NULL CHECK (t >= 0), ctx TEXT NOT NULL, lamport INTEGER NOT NULL, sync_state TEXT NOT NULL DEFAULT 'local',\n",
        "                  checksum TEXT NOT NULL, CHECK (op = 'adjust' OR qty > 0));\n",
        "CREATE TABLE meta_edit (id TEXT PRIMARY KEY, item TEXT NOT NULL REFERENCES item(item), field TEXT NOT NULL, value TEXT NOT NULL, node TEXT NOT NULL, lamport INTEGER NOT NULL);\n",
        "CREATE TABLE conflict (id INTEGER PRIMARY KEY AUTOINCREMENT, item TEXT NOT NULL, field TEXT NOT NULL, a TEXT, b TEXT, status TEXT NOT NULL DEFAULT 'PENDING');\n",
        "CREATE TABLE audit (seq INTEGER PRIMARY KEY AUTOINCREMENT, event TEXT NOT NULL, who TEXT, detail TEXT, prev_hash TEXT, hash TEXT);\n",
        "CREATE VIEW stock AS SELECT item, SUM(CASE WHEN op='out' THEN -qty ELSE qty END) AS delta FROM txn GROUP BY item;\n",
        "\"\"\"\n",
        "db = sqlite3.connect(\":memory:\")\n",
        "db.executescript(DDL)\n",
        "db.executemany(\"INSERT INTO item VALUES (?,?,?,?,?)\", [(x[\"item\"], x[\"cat\"], x[\"supplier\"], x[\"lead_days\"], x[\"min_level\"]) for x in ITEMS])\n",
        "with db:\n",
        "    db.executemany(\"INSERT INTO txn VALUES (?,?,?,?,?,?,?,?,?,?)\", [(r[\"id\"], r[\"node\"], r[\"item\"], r[\"op\"], r[\"qty\"], r[\"t\"], r[\"ctx\"], r[\"lamport\"], r[\"sync_state\"], r[\"checksum\"]) for r in recs])\n",
        "n_rows = db.execute(\"SELECT COUNT(*) FROM txn\").fetchone()[0]\n",
        "print(f\"\\n[로컬 DB] 테이블 5개·뷰 1개 생성, 트랜잭션 {n_rows}건 적재 (스키마 JSON과 행 수 일치: {n_rows == len(recs)})\")\n",
        "tests = {\"중복 ID 입력 차단(멱등 키)\": (\"INSERT INTO txn VALUES (?,?,?,?,?,?,?,?,?,?)\", (r0[\"id\"], r0[\"node\"], r0[\"item\"], \"in\", 5, 1.0, \"평시\", 1, \"local\", \"abcd1234\")),\n",
        "         \"존재하지 않는 품목 차단(외래 키)\": (\"INSERT INTO txn VALUES ('보급소A-99999','보급소A','I99','in',5,1.0,'평시',1,'local','abcd1234')\", ()),\n",
        "         \"불출 0개 차단(CHECK)\": (\"INSERT INTO txn VALUES ('보급소A-99998','보급소A','I01','out',0,1.0,'평시',1,'local','abcd1234')\", ()),\n",
        "         \"음수 시각 차단(CHECK)\": (\"INSERT INTO txn VALUES ('보급소A-99997','보급소A','I01','in',5,-1.0,'평시',1,'local','abcd1234')\", ()),\n",
        "         \"허용되지 않은 op 차단(CHECK)\": (\"INSERT INTO txn VALUES ('보급소A-99996','보급소A','I01','sell',5,1.0,'평시',1,'local','abcd1234')\", ())}\n",
        "db.execute(\"PRAGMA foreign_keys = ON\")\n",
        "blocked = 0\n",
        "for n, (sql, args) in tests.items():\n",
        "    try:\n",
        "        with db:\n",
        "            db.execute(sql, args)\n",
        "        print(f\"  {n:28s}-> 통과(결함)\")\n",
        "    except sqlite3.IntegrityError as e:\n",
        "        blocked += 1; print(f\"  {n:28s}-> 차단: {str(e)[:48]}\")\n",
        "print(f\"제약 조건 시험 {blocked}/{len(tests)} 차단, 롤백 후 행 수 {db.execute('SELECT COUNT(*) FROM txn').fetchone()[0]}건 유지\")\n",
        "\n",
        "# 4) 요구사항 12개와 검증 결과(코드 시험으로 확인된 항목은 PASS)\n",
        "names = {r[0] for r in db.execute(\"SELECT name FROM sqlite_master\")}\n",
        "cols = {r[1] for r in db.execute(\"PRAGMA table_info(txn)\")}\n",
        "REQ = [(\"R01\", \"가용성\", \"통신이 끊겨도 단말 로컬 DB에 입고·불출을 기록할 수 있다\", \"메모리 DB 적재\", n_rows == len(recs)), (\"R02\", \"무결성\", \"같은 입력이 두 번 들어와도 한 번만 반영한다\", \"ID 중복 차단\", blocked >= 1),\n",
        "       (\"R03\", \"무결성\", \"불출 수량 0·잘못된 구분 값은 저장되지 않는다\", \"CHECK 제약\", blocked >= 3), (\"R04\", \"무결성\", \"존재하지 않는 품목은 입력할 수 없다\", \"외래 키\", blocked >= 2),\n",
        "       (\"R05\", \"무결성\", \"레코드마다 위변조 점검 해시를 둔다\", \"checksum 필드\", all(re.match(r\"^[0-9a-f]{8}$\", r[\"checksum\"]) for r in recs)), (\"R06\", \"동기화\", \"동기화 상태(local/sent/acked)를 레코드에 기록한다\", \"sync_state 필드\", \"sync_state\" in cols),\n",
        "       (\"R07\", \"동기화\", \"노드별 논리 시계로 선후 관계를 비교한다\", \"lamport 필드\", all(r[\"lamport\"] > 0 for r in recs)), (\"R08\", \"동기화\", \"같은 필드를 동시에 고친 충돌은 보류 목록에 남긴다\", \"conflict 테이블\", \"conflict\" in names),\n",
        "       (\"R09\", \"보안\", \"변경·동기화·승인 행위는 해시가 이어진 감사 로그에 남긴다\", \"audit 테이블\", \"audit\" in names), (\"R10\", \"보안\", \"개인 식별 정보는 입력하지 않는다\", \"note 점검\", caught == len(inj)),\n",
        "       (\"R11\", \"운영\", \"입력 오류는 사유와 함께 즉시 알려 준다\", \"검증 오류 메시지\", caught >= 6), (\"R12\", \"운영\", \"동기화 후 재고를 업체·품목별로 마이닝해 공급망 재난 징후를 식별한다\", \"stock 뷰\", db.execute(\"SELECT COUNT(*) FROM stock\").fetchone()[0] == 12)]\n",
        "print(\"\\n[요구사항 검증표]\")\n",
        "for i, c, d, m, ok in REQ:\n",
        "    print(f\"  {i} [{c}] {d[:34]:36s} {m:12s} {'PASS' if ok else 'FAIL'}\")\n",
        "print(f\"요구사항 {sum(x[4] for x in REQ)}/{len(REQ)} 충족\")\n",
        "js = json.dumps({\"title\": \"dat10_input_schema\", \"fields\": {k: {kk: vv for kk, vv in v.items() if kk != 'desc'} for k, v in list(SCHEMA.items())[:2]}}, ensure_ascii=False)\n",
        "print(\"스키마 JSON(앞 2개 필드):\", js[:140], \"...\")\n",
        "print(\"[입출력] 입력: 보급소 트랜잭션(12필드) / 출력: 동기화된 재고, 충돌 목록, 공급망 재난 징후, 감사 로그\")\n",
        "print(\"※ 가상 자료이며 실제 재고·업체 정보는 사용하지 않는다. 실제 단말 DB는 암호화 저장 등 보안 요건을 별도로 충족해야 한다.\")\n",
        "\n",
        "st = stock_of(txns)\n",
        "plt.figure(figsize=(8, 3.8)); plt.bar(list(st), list(st.values()), color=\"tab:blue\"); plt.title(\"Synthetic stock by item after 7 days (all nodes merged)\"); plt.xlabel(\"item\"); plt.ylabel(\"units\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[스키마] 필드 12개, type·unit·desc 모두 정의된 필드 12개 (정의 완성도 100%)\n",
        "[정상 자료 검증] 트랜잭션 2100건 중 오류 0건 (목표 오류 0건: 달성), 개인 식별 필드 0개\n",
        "\n",
        "[오류 주입 시험]\n",
        "  자료형 오류(qty 문자)        -> 검출: qty: 자료형 오류(str)\n",
        "  허용값 오류(품목)            -> 검출: item: 허용되지 않은 값 'I99'\n",
        "  형식 오류(ID)             -> 검출: id: 형식 불일치 'A-1'\n",
        "  범위 오류(시각)             -> 검출: t: 최소값 0 미만\n",
        "  논리 오류(불출 0개)          -> 검출: qty: 입고·불출 수량은 1 이상\n",
        "  논리 오류(ID 접두어)         -> 검출: id: 노드 접두어와 node 불일치\n",
        "  필드 누락                 -> 검출: checksum: 필수 값 누락\n",
        "  개인정보 의심 문구            -> 검출: note: 개인 식별 정보 의심 문구\n",
        "주입 오류 8종 중 8종 검출\n",
        "\n",
        "[로컬 DB] 테이블 5개·뷰 1개 생성, 트랜잭션 2100건 적재 (스키마 JSON과 행 수 일치: True)\n",
        "  중복 ID 입력 차단(멱등 키)           -> 차단: UNIQUE constraint failed: txn.id\n",
        "  존재하지 않는 품목 차단(외래 키)         -> 차단: FOREIGN KEY constraint failed\n",
        "  불출 0개 차단(CHECK)             -> 차단: CHECK constraint failed: op = 'adjust' OR qty > \n",
        "  음수 시각 차단(CHECK)             -> 차단: CHECK constraint failed: t >= 0\n",
        "  허용되지 않은 op 차단(CHECK)        -> 차단: CHECK constraint failed: op IN ('in','out','adju\n",
        "제약 조건 시험 5/5 차단, 롤백 후 행 수 2100건 유지\n",
        "\n",
        "[요구사항 검증표]\n",
        "  R01 [가용성] 통신이 끊겨도 단말 로컬 DB에 입고·불출을 기록할 수 있다    메모리 DB 적재    PASS\n",
        "  R02 [무결성] 같은 입력이 두 번 들어와도 한 번만 반영한다            ID 중복 차단     PASS\n",
        "  R03 [무결성] 불출 수량 0·잘못된 구분 값은 저장되지 않는다           CHECK 제약     PASS\n",
        "  R04 [무결성] 존재하지 않는 품목은 입력할 수 없다                 외래 키         PASS\n",
        "  R05 [무결성] 레코드마다 위변조 점검 해시를 둔다                  checksum 필드  PASS\n",
        "  R06 [동기화] 동기화 상태(local/sent/acked)를 레코드에 기록한   sync_state 필드 PASS\n",
        "  R07 [동기화] 노드별 논리 시계로 선후 관계를 비교한다               lamport 필드   PASS\n",
        "  R08 [동기화] 같은 필드를 동시에 고친 충돌은 보류 목록에 남긴다         conflict 테이블 PASS\n",
        "  R09 [보안] 변경·동기화·승인 행위는 해시가 이어진 감사 로그에 남긴다     audit 테이블    PASS\n",
        "  R10 [보안] 개인 식별 정보는 입력하지 않는다                   note 점검      PASS\n",
        "  R11 [운영] 입력 오류는 사유와 함께 즉시 알려 준다               검증 오류 메시지    PASS\n",
        "  R12 [운영] 동기화 후 재고를 업체·품목별로 마이닝해 공급망 재난 징후를    stock 뷰      PASS\n",
        "요구사항 12/12 충족\n",
        "스키마 JSON(앞 2개 필드): {\"title\": \"dat10_input_schema\", \"fields\": {\"id\": {\"type\": \"string\", \"pattern\": \"^보급소[A-E]-\\\\d{5}$\", \"unit\": \"-\"}, \"node\": {\"type\": \"string\", ...\n",
        "[입출력] 입력: 보급소 트랜잭션(12필드) / 출력: 동기화된 재고, 충돌 목록, 공급망 재난 징후, 감사 로그\n",
        "※ 가상 자료이며 실제 재고·업체 정보는 사용하지 않는다. 실제 단말 DB는 암호화 저장 등 보안 요건을 별도로 충족해야 한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 02. 오프라인 데이터 입력 유효성 검증 및 로컬 SQLite 트랜잭션 전처리 모듈 구현\n",
        "**소요시간**: 약 2.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "오프라인 데이터 입력 유효성 검증 및 로컬 트랜잭션 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **형식이 섞인 오프라인 입력 줄을 파싱·검증하고 SQLite 트랜잭션으로 안전하게 적재하는 전처리 모듈을 구현해 처리 속도 1,000건/초 이상, 전처리 오류율 0%를 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **ID\\ 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/preprocess/dat10_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": [
        "\"\"\"서식이 서로 다른 오프라인 입력 문자열 8,000줄(문장형·CSV·JSON·키=값 4종)을 파싱·검증해 SQLite 메모리 DB에 배치 트랜잭션으로 적재하고, 입력 구분(입고·불출·조정)별 동적 임계값으로 이상 수량을 격리하며 중복 재전송 멱등 처리·배치 실패 롤백·처리 속도(건/초)·정답 대조 오류율을 측정한다.\"\"\"\n",
        "import json\n",
        "import re\n",
        "import sqlite3\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 오프라인 입력 원문 생성: 같은 트랜잭션을 서식 4종 중 하나로 적고, 잡음(이상 수량·형식 깨짐·재전송 중복)을 섞는다.\n",
        "# ---------------------------------------------------------------------------\n",
        "rng = np.random.default_rng(21)\n",
        "base = make_txns(31, 7, per_node_day=230)                      # 5노드 x 7일 x 230건 = 8,050건\n",
        "OPK = {\"in\": \"입고\", \"out\": \"불출\", \"adjust\": \"조정\"}\n",
        "lines, truth = [], {}\n",
        "for i, x in enumerate(base):\n",
        "    f = i % 4\n",
        "    if f == 0:\n",
        "        s = f\"[입력] {x['node']} 품목{x['item']} {OPK[x['op']]} {x['qty']}개 ({x['ctx']}) {x['t']}h #{x['id']}\"\n",
        "    elif f == 1:\n",
        "        s = f\"{x['node']},{x['item']},{x['op']},{x['qty']},{x['t']},{x['ctx']},{x['id']}\"\n",
        "    elif f == 2:\n",
        "        s = json.dumps({k: x[k] for k in (\"node\", \"item\", \"op\", \"qty\", \"t\", \"ctx\", \"id\")}, ensure_ascii=False)\n",
        "    else:\n",
        "        s = f\"품목 {x['item']} / {OPK[x['op']]} / {x['qty']}개 / {x['node']} / t={x['t']} / ctx={x['ctx']} / id={x['id']}\"\n",
        "    lines.append(s); truth[x[\"id\"]] = x\n",
        "noise_ids = set()\n",
        "for i in rng.choice(len(lines), 120, replace=False):          # 잡음 주입: 이상 수량 40, 깨진 줄 40, 존재하지 않는 품목 40\n",
        "    x = base[i]\n",
        "    if i % 3 == 0:\n",
        "        lines[i] = lines[i].replace(str(x[\"qty\"]), \"9999\", 1) if x[\"op\"] != \"adjust\" else lines[i]; noise_ids.add(x[\"id\"]) if \"9999\" in lines[i] else None\n",
        "    elif i % 3 == 1:\n",
        "        lines[i] = lines[i][: len(lines[i]) // 2]; noise_ids.add(x[\"id\"])\n",
        "    else:\n",
        "        lines[i] = lines[i].replace(x[\"item\"], \"I99\"); noise_ids.add(x[\"id\"])\n",
        "dups = [lines[i] for i in rng.choice(len(lines), 200, replace=False)]   # 통신 복구 후 재전송된 중복 줄 200개\n",
        "lines = lines + dups\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# 파서: 서식별 정규식 -> 표준 레코드 (해석 실패는 예외)\n",
        "# ---------------------------------------------------------------------------\n",
        "IV = {v: k for k, v in OPK.items()}\n",
        "P1 = re.compile(r\"\\[입력\\]\\s*(보급소[A-E])\\s*품목(I\\d{2})\\s*(입고|불출|조정)\\s*(-?\\d+)개\\s*\\((\\S+)\\)\\s*([\\d.]+)h\\s*#(\\S+)\")\n",
        "P4 = re.compile(r\"품목\\s*(I\\d{2})\\s*/\\s*(입고|불출|조정)\\s*/\\s*(-?\\d+)개\\s*/\\s*(보급소[A-E])\\s*/\\s*t=([\\d.]+)\\s*/\\s*ctx=(\\S+)\\s*/\\s*id=(\\S+)\")\n",
        "\n",
        "\n",
        "def parse(s):\n",
        "    s = s.strip()\n",
        "    if s.startswith(\"[\"):\n",
        "        m = P1.match(s)\n",
        "        if m:\n",
        "            return {\"node\": m[1], \"item\": m[2], \"op\": IV[m[3]], \"qty\": int(m[4]), \"ctx\": m[5], \"t\": float(m[6]), \"id\": m[7]}\n",
        "    elif s.startswith(\"{\"):\n",
        "        return json.loads(s)\n",
        "    elif s.startswith(\"품목\"):\n",
        "        m = P4.match(s)\n",
        "        if m:\n",
        "            return {\"item\": m[1], \"op\": IV[m[2]], \"qty\": int(m[3]), \"node\": m[4], \"t\": float(m[5]), \"ctx\": m[6], \"id\": m[7]}\n",
        "    else:\n",
        "        p = s.split(\",\")\n",
        "        if len(p) == 7:\n",
        "            return {\"node\": p[0], \"item\": p[1], \"op\": p[2], \"qty\": int(p[3]), \"t\": float(p[4]), \"ctx\": p[5], \"id\": p[6]}\n",
        "    raise ValueError(\"서식 해석 불가\")\n",
        "\n",
        "\n",
        "def valid(r):\n",
        "    if r[\"item\"] not in ITEM_IDS or r[\"node\"] not in NODES or r[\"op\"] not in OPK or r[\"ctx\"] not in CTXS:\n",
        "        return \"허용값 오류\"\n",
        "    if r[\"op\"] != \"adjust\" and r[\"qty\"] <= 0:\n",
        "        return \"수량 오류\"\n",
        "    if not re.match(r\"^보급소[A-E]-\\d{5}$\", r[\"id\"]) or not r[\"id\"].startswith(r[\"node\"]):\n",
        "        return \"ID 오류\"\n",
        "    return None\n",
        "\n",
        "\n",
        "# 1) 파싱 + 검증 + 동적 임계값\n",
        "t0 = time.perf_counter()\n",
        "parsed, rejected = [], []\n",
        "for s in lines:\n",
        "    try:\n",
        "        r = parse(s); e = valid(r)\n",
        "        (rejected if e else parsed).append((r, e) if e else r)\n",
        "    except (ValueError, KeyError, json.JSONDecodeError, IndexError) as ex:\n",
        "        rejected.append((s, \"해석 실패\"))\n",
        "df = pd.DataFrame(parsed)\n",
        "before = len(df)\n",
        "df = df.drop_duplicates(\"id\")\n",
        "th = {}\n",
        "for op, g in df[df[\"op\"] != \"adjust\"].groupby(\"op\"):\n",
        "    med = g[\"qty\"].median(); mad = (g[\"qty\"] - med).abs().median() * 1.4826; th[op] = (max(med - 6 * mad, 1), med + 6 * mad)\n",
        "fixed = (df[\"op\"] != \"adjust\") & (df[\"qty\"] > 100)\n",
        "dyn = pd.Series([(op != \"adjust\") and not (th[op][0] <= q <= th[op][1]) for op, q in zip(df[\"op\"], df[\"qty\"])], index=df.index)\n",
        "susp = dyn.to_numpy()\n",
        "clean = df[~dyn].copy()\n",
        "dt_parse = time.perf_counter() - t0\n",
        "rej_cnt = pd.Series([e for _, e in rejected]).value_counts().to_dict()\n",
        "print(f\"[파싱·검증] 입력 {len(lines)}줄(재전송 중복 200 포함): 해석 실패·허용값 오류 {len(rejected)}줄 {rej_cnt}, 통과 {before}줄, 중복 ID {before - len(df)}줄 제거\")\n",
        "print(\"구분별 동적 임계값(수량):\", \", \".join(f\"{OPK[k]} {lo:.0f}~{hi:.0f}\" for k, (lo, hi) in th.items()))\n",
        "true_bad = df[\"id\"].isin(noise_ids).to_numpy()\n",
        "for name, fl in [(\"고정 임계값(100개 초과)\", fixed.to_numpy()), (\"구분별 동적 임계값\", susp)]:\n",
        "    print(f\"  {name:20s}: 이상 판정 {int(fl.sum())}건, 이상 수량 {int((fl & true_bad).sum())}/{int(true_bad.sum())} 검출, 정상 입고인데 이상 판정(오탐) {int((fl & ~true_bad).sum())}건\")\n",
        "\n",
        "# 2) SQLite 배치 트랜잭션 적재(멱등: 같은 ID는 한 번만), 배치 실패 시 롤백 후 행 단위로 격리\n",
        "db = sqlite3.connect(\":memory:\")\n",
        "db.execute(\"CREATE TABLE txn (id TEXT PRIMARY KEY, node TEXT NOT NULL, item TEXT NOT NULL, op TEXT NOT NULL CHECK (op IN ('in','out','adjust')), qty INTEGER NOT NULL, t REAL NOT NULL CHECK (t >= 0), ctx TEXT NOT NULL)\")\n",
        "db.execute(\"CREATE TABLE quarantine (id TEXT, reason TEXT)\")\n",
        "rows = [(r[\"id\"], r[\"node\"], r[\"item\"], r[\"op\"], int(r[\"qty\"]), float(r[\"t\"]), r[\"ctx\"]) for r in clean.to_dict(\"records\")]\n",
        "rows[777] = (rows[777][0], rows[777][1], rows[777][2], \"sell\", 5, 1.0, \"평시\")      # 배치 안에 제약 위반 1건 주입\n",
        "t1 = time.perf_counter(); inserted = rolled = 0\n",
        "for b in range(0, len(rows), 500):\n",
        "    batch = rows[b:b + 500]\n",
        "    try:\n",
        "        with db:\n",
        "            db.executemany(\"INSERT INTO txn VALUES (?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING\", batch)\n",
        "        inserted += len(batch)\n",
        "    except sqlite3.IntegrityError:\n",
        "        rolled += 1                                            # 배치 전체 롤백 후 행 단위 재시도\n",
        "        for r in batch:\n",
        "            try:\n",
        "                with db:\n",
        "                    db.execute(\"INSERT INTO txn VALUES (?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING\", r)\n",
        "                inserted += 1\n",
        "            except sqlite3.IntegrityError as e:\n",
        "                with db:\n",
        "                    db.execute(\"INSERT INTO quarantine VALUES (?,?)\", (r[0], \"제약 위반\"))\n",
        "dt_db = time.perf_counter() - t1\n",
        "n_db = db.execute(\"SELECT COUNT(*) FROM txn\").fetchone()[0]\n",
        "print(f\"\\n[SQLite 적재] 500건 배치 {len(rows) // 500 + 1}개: 롤백 {rolled}회(제약 위반 1건 -> 행 단위 격리), 적재 {n_db}건, 격리 테이블 {db.execute('SELECT COUNT(*) FROM quarantine').fetchone()[0]}건\")\n",
        "db.execute(\"INSERT INTO txn VALUES (?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING\", rows[0]); n2 = db.execute(\"SELECT COUNT(*) FROM txn\").fetchone()[0]\n",
        "print(f\"같은 ID를 다시 넣어도 행 수 {n2}건 유지(멱등)\")\n",
        "\n",
        "# 3) 정답 대조: 적재된 행의 필드가 원본 트랜잭션과 같은가, 속도\n",
        "errs = 0; chk = 0\n",
        "for (i, node, item, op, qty, t, ctx) in db.execute(\"SELECT * FROM txn\"):\n",
        "    x = truth[i]; chk += 1\n",
        "    errs += (x[\"node\"], x[\"item\"], x[\"op\"], x[\"qty\"], x[\"ctx\"]) != (node, item, op, qty, ctx) or abs(x[\"t\"] - t) > 1e-6\n",
        "total = dt_parse + dt_db\n",
        "print(f\"\\n[전처리 정확도] 적재 {chk}건 중 원본과 불일치 {errs}건 -> 전처리 오류율 {errs / chk:.2%} (목표 0%: {'달성' if errs == 0 else '미달'})\")\n",
        "print(f\"처리 속도 {len(lines) / total:,.0f}건/초 (파싱·검증 {dt_parse:.2f}초 + DB 적재 {dt_db:.2f}초, 목표 1,000건/초 이상: {'달성' if len(lines) / total >= 1000 else '미달'})\")\n",
        "print(f\"원본 {len(base)}건 중 잡음 주입 {len(noise_ids)}건과 격리 {int(susp.sum())}건을 제외하고 적재 {n_db}건 -> 격리된 건은 담당자가 원본 확인 후 재입력\")\n",
        "print(\"※ 이상 판정된 줄은 삭제하지 않고 격리하며, 임계값은 입력 구분별 중앙값±6MAD로 자료에서 조정한다. 실제 서식에 맞는 정규식은 현장에서 보완한다.\")\n",
        "\n",
        "plt.figure(figsize=(8, 4))\n",
        "for op, g in df[df[\"op\"] != \"adjust\"].groupby(\"op\"):\n",
        "    plt.hist(g[\"qty\"].clip(upper=200), bins=40, alpha=.5, label=op)\n",
        "plt.axvline(100, color=\"k\", ls=\"--\"); plt.legend(); plt.xlabel(\"qty\"); plt.title(\"Quantity by operation (dashed: fixed threshold 100)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[파싱·검증] 입력 8250줄(재전송 중복 200 포함): 해석 실패·허용값 오류 84줄 {'허용값 오류': 44, '해석 실패': 40}, 통과 8166줄, 중복 ID 198줄 제거\n",
        "구분별 동적 임계값(수량): 입고 1~221, 불출 1~56\n",
        "  고정 임계값(100개 초과)     : 이상 판정 139건, 이상 수량 34/34 검출, 정상 입고인데 이상 판정(오탐) 105건\n",
        "  구분별 동적 임계값          : 이상 판정 34건, 이상 수량 34/34 검출, 정상 입고인데 이상 판정(오탐) 0건\n",
        "\n",
        "[SQLite 적재] 500건 배치 16개: 롤백 1회(제약 위반 1건 -> 행 단위 격리), 적재 7933건, 격리 테이블 1건\n",
        "같은 ID를 다시 넣어도 행 수 7933건 유지(멱등)\n",
        "\n",
        "[전처리 정확도] 적재 7933건 중 원본과 불일치 0건 -> 전처리 오류율 0.00% (목표 0%: 달성)\n",
        "처리 속도 48,683건/초 (파싱·검증 0.13초 + DB 적재 0.04초, 목표 1,000건/초 이상: 달성)\n",
        "원본 8050건 중 잡음 주입 116건과 격리 34건을 제외하고 적재 7933건 -> 격리된 건은 담당자가 원본 확인 후 재입력\n",
        "※ 이상 판정된 줄은 삭제하지 않고 격리하며, 임계값은 입력 구분별 중앙값±6MAD로 자료에서 조정한다. 실제 서식에 맞는 정규식은 현장에서 보완한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 03. 오프라인 단말 간 1차 피어투피어(P2P)/USB 재고 데이터 교환 Baseline 기능 구현\n",
        "**소요시간**: 약 2.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "오프라인 단말 간 피어투피어(P2P) 재고 데이터 교환 Baseline 기능을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **보급소 단말 5곳이 접촉할 때 USB·P2P로 재고 트랜잭션을 주고받는 Baseline 교환 기능을 만들고 7일 후 정보 공유 정확도 75% 이상을 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **7일간 접촉 56회(접촉당 방향별 최대 80건, 손실 5%)의 가상 일정에서 상대 상태를 모르는 전체 전송, ID 차집합 델타 전송, 델타에 우선순위(오래된 것 먼저)를 더한 방식을 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/models/dat10_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": [
        "\"\"\"오프라인 보급소 5곳이 끊긴 통신 대신 짧은 P2P 접촉·USB 인편으로 트랜잭션을 교환하는 Baseline을 구현해, 전체 전송·ID 차집합(델타) 전송·델타+우선순위 휴리스틱(내 기록 우선·오래된 것 우선)의 7일 후 동기화 완전성·전송 효율·수렴 시점을 비교하고 우선순위 가중치를 1차 튜닝한다.\"\"\"\n",
        "import itertools\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "TX = make_txns(7, 7, per_node_day=60)                          # 5노드 x 7일 x 60건 = 2,100건\n",
        "N = len(TX); IDX = {x[\"id\"]: i for i, x in enumerate(TX)}\n",
        "ORIGIN = np.array([NODES.index(x[\"node\"]) for x in TX]); T = np.array([x[\"t\"] for x in TX])\n",
        "CONTACTS_PER_DAY, BW, LOSS = 8, 80, 0.05                       # 하루 접촉 8회, 접촉당 방향별 최대 60건, 전송 손실 5%\n",
        "\n",
        "\n",
        "def contact_schedule(seed=3, days=7):\n",
        "    r = np.random.default_rng(seed); out = []\n",
        "    for d in range(days):\n",
        "        for _ in range(CONTACTS_PER_DAY):\n",
        "            a, b = r.choice(5, 2, replace=False); out.append((d * 24 + float(r.uniform(0, 24)), int(a), int(b), r.random()))\n",
        "    return sorted(out)\n",
        "\n",
        "\n",
        "def simulate(method, w_own=0.0, w_age=0.0, seed=3, trace=False):\n",
        "    r = np.random.default_rng(seed + 100)\n",
        "    known = [set() for _ in range(5)]; sent = useful = 0; tr = []\n",
        "    prod = [np.where(ORIGIN == n)[0] for n in range(5)]\n",
        "    for tm, a, b, _ in contact_schedule(seed):\n",
        "        for n in (a, b):                                       # 접촉 시점까지 자기 노드가 기록한 트랜잭션\n",
        "            known[n] |= set(prod[n][T[prod[n]] <= tm].tolist())\n",
        "        for src, dst in ((a, b), (b, a), ):\n",
        "            if method == \"full\":\n",
        "                cand = sorted(known[src], key=lambda i: T[i])    # 상대가 뭘 가졌는지 모르고 전부 보냄(앞에서부터 BW건)\n",
        "            else:\n",
        "                cand = [i for i in known[src] if i not in known[dst]]        # 상대가 없는 것만(ID 요약 비교)\n",
        "                if method == \"prio\":\n",
        "                    cand.sort(key=lambda i: -(w_own * (ORIGIN[i] == src) + w_age * (tm - T[i]) / 168))\n",
        "                else:\n",
        "                    cand.sort(key=lambda i: T[i])\n",
        "            for i in cand[:BW]:\n",
        "                sent += 1\n",
        "                if r.random() > LOSS:\n",
        "                    useful += i not in known[dst]; known[dst].add(i)\n",
        "        if trace:\n",
        "            tr.append((tm, np.mean([len(k) for k in known]) / N))\n",
        "    comp = float(np.mean([len(k) for k in known]) / N)\n",
        "    return {\"comp\": comp, \"sent\": sent, \"useful\": useful, \"eff\": useful / max(sent, 1), \"full_nodes\": sum(len(k) == N for k in known), \"trace\": tr}\n",
        "\n",
        "\n",
        "# 1) 방식 비교\n",
        "res = {\"전체 전송(상대 상태 모름)\": simulate(\"full\"), \"ID 차집합 델타(오래된 것 먼저)\": simulate(\"delta\"), \"델타 + 우선순위(기본 가중 own 1, age 0)\": simulate(\"prio\", 1, 0)}\n",
        "print(f\"[조건] 보급소 5곳, 트랜잭션 {N}건, 7일간 접촉 {CONTACTS_PER_DAY * 7}회(P2P·USB), 접촉당 방향별 최대 {BW}건, 손실 {LOSS:.0%}\")\n",
        "print(f\"{'방식':32s} {'완전성':>7s} {'전송':>6s} {'유효 전송':>8s} {'효율':>6s}\")\n",
        "for k, v in res.items():\n",
        "    print(f\"{k:32s} {v['comp']:7.1%} {v['sent']:6d} {v['useful']:8d} {v['eff']:6.1%}\")\n",
        "\n",
        "# 2) 1차 튜닝: 우선순위 가중치 격자 탐색(스케줄 시드 3개 평균, 시험은 다른 시드 3개)\n",
        "grid = {}\n",
        "for wo, wa in itertools.product((0, 1, 2, 4), (-1, 0, 1)):\n",
        "    grid[(wo, wa)] = np.mean([simulate(\"prio\", wo, wa, seed=s)[\"comp\"] for s in (1, 2, 4)])\n",
        "best = max(grid, key=grid.get)\n",
        "print(f\"\\n[우선순위 가중치 탐색(튜닝 시드 3개 평균 완전성)] 내 기록 가중 x 오래됨 가중\")\n",
        "print(\"  \" + \" \".join(f\"own{wo}/age{wa}:{v:.1%}\" for (wo, wa), v in list(grid.items())[:6]) + \" ...\")\n",
        "print(f\"채택: own {best[0]}, age {best[1]} (튜닝 평균 {grid[best]:.1%})\")\n",
        "test = {m: np.mean([simulate(m, *( (best if m == 'prio' else (0, 0)) ), seed=s)[\"comp\"] for s in (5, 6, 7)]) for m in (\"full\", \"delta\", \"prio\")}\n",
        "print(f\"시험 시드 3개 평균 완전성: 전체 전송 {test['full']:.1%}, 델타 {test['delta']:.1%}, 델타+튜닝 우선순위 {test['prio']:.1%}\")\n",
        "acc = test[\"prio\"]\n",
        "print(f\"기본 정확도(7일 후 모든 노드가 전체 트랜잭션 중 알고 있는 비율) {acc:.1%} (목표 75% 이상: {'달성' if acc >= 0.75 else '미달'})\")\n",
        "\n",
        "# 3) 수렴 추적과 판단 지연\n",
        "import time\n",
        "t0 = time.perf_counter(); r = simulate(\"prio\", *best, trace=True); dt = time.perf_counter() - t0\n",
        "tr = r[\"trace\"]; day = next((t / 24 for t, c in tr if c >= 0.9), None)\n",
        "print(f\"\\n[수렴] 7일 시뮬레이션 {dt * 1000:.0f}ms (접촉 1회당 {dt / len(tr) * 1000:.1f}ms), 완전성 90% 도달 시점 {('%.1f일' % day) if day else '7일 내 미도달'}, 모든 트랜잭션을 가진 노드 {r['full_nodes']}/5\")\n",
        "print(\"접촉이 부족하면 완전성이 100%에 못 미치므로, 동기화 전 재고는 '확인된 범위' 표시가 필요하다(M6).\")\n",
        "print(\"※ 가상 접촉 일정과 대역폭 가정에 따른 결과이며, 실제 접촉 빈도·USB 이동은 현장 조건에 따라 다르다. 병합 충돌 판단은 M4·M7에서 다룬다.\")\n",
        "\n",
        "plt.figure(figsize=(8, 4))\n",
        "for m, lab in ((\"full\", \"full dump\"), (\"delta\", \"delta\"), (\"prio\", \"delta + priority\")):\n",
        "    rr = simulate(m, *(best if m == \"prio\" else (0, 0)), trace=True)[\"trace\"]; plt.plot([t / 24 for t, _ in rr], [c * 100 for _, c in rr], label=lab)\n",
        "plt.xlabel(\"days\"); plt.ylabel(\"mean completeness (%)\"); plt.title(\"P2P/USB exchange: completeness over time\"); plt.legend(); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[조건] 보급소 5곳, 트랜잭션 2100건, 7일간 접촉 56회(P2P·USB), 접촉당 방향별 최대 80건, 손실 5%\n",
        "방식                                   완전성     전송    유효 전송     효율\n",
        "전체 전송(상대 상태 모름)                    22.8%   8470      398   4.7%\n",
        "ID 차집합 델타(오래된 것 먼저)                82.5%   6999     6657  95.1%\n",
        "델타 + 우선순위(기본 가중 own 1, age 0)      83.7%   7143     6791  95.1%\n",
        "\n",
        "[우선순위 가중치 탐색(튜닝 시드 3개 평균 완전성)] 내 기록 가중 x 오래됨 가중\n",
        "  own0/age-1:84.8% own0/age0:83.7% own0/age1:82.7% own1/age-1:84.7% own1/age0:84.4% own1/age1:84.0% ...\n",
        "채택: own 0, age -1 (튜닝 평균 84.8%)\n",
        "시험 시드 3개 평균 완전성: 전체 전송 24.0%, 델타 82.1%, 델타+튜닝 우선순위 84.6%\n",
        "기본 정확도(7일 후 모든 노드가 전체 트랜잭션 중 알고 있는 비율) 84.6% (목표 75% 이상: 달성)\n",
        "\n",
        "[수렴] 7일 시뮬레이션 58ms (접촉 1회당 1.0ms), 완전성 90% 도달 시점 7일 내 미도달, 모든 트랜잭션을 가진 노드 0/5\n",
        "접촉이 부족하면 완전성이 100%에 못 미치므로, 동기화 전 재고는 '확인된 범위' 표시가 필요하다(M6).\n",
        "※ 가상 접촉 일정과 대역폭 가정에 따른 결과이며, 실제 접촉 빈도·USB 이동은 현장 조건에 따라 다르다. 병합 충돌 판단은 M4·M7에서 다룬다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 04. 네트워크 복구 시 클라우드 중앙 서버와의 양방향 데이터 충돌 해결(CRDT) 동기화 고도화\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "네트워크 복구 시 중앙 서버와의 양방향 데이터 충돌 해결(CRDT) 동기화를 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **통신 복구 후 노드와 중앙 서버가 양방향으로 병합하는 CRDT 동기화를 구현하고, 같은 품목 정보를 서로 모르게 고친 충돌의 탐지 재현율 85% 이상과 처리 지연 200ms 이하를 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **재고 수량은 품목별 PN-카운터(노드별 증가·감소 누적, 병합은 요소별 최댓값)로 만들어 노드 5곳과 서버를 2단계로 병합하고, 마지막 값이 이기는 방식(LWW)과 갱신 유실을 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/models/dat10_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": [
        "\"\"\"통신 복구 후 보급소 5곳과 중앙 서버가 양방향으로 병합하는 CRDT 동기화(재고는 PN-카운터, 메타 정보는 버전 벡터 기반 충돌 감지)를 구현해, 수렴·갱신 유실 0건·멱등/교환/결합 성질·중복 재전송·오래된 상태 재전송 방어를 검증하고, 충돌 탐지 재현율(시간 비교 규칙 대 버전 벡터 규칙)과 오경보 패턴·동기화 지연을 측정한다.\"\"\"\n",
        "import copy\n",
        "import itertools\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "\n",
        "class Replica:\n",
        "    \"\"\"재고는 품목별 PN-카운터(노드별 증가량 P, 감소량 N), 같은 노드의 기록은 서로 덮어쓰지 않는다. 병합은 요소별 최댓값.\"\"\"\n",
        "    def __init__(self, name):\n",
        "        self.name = name; self.P = {i: {} for i in ITEM_IDS}; self.N = {i: {} for i in ITEM_IDS}\n",
        "\n",
        "    def apply(self, x):\n",
        "        d = self.P if x[\"op\"] == \"in\" or (x[\"op\"] == \"adjust\" and x[\"qty\"] > 0) else self.N\n",
        "        d[x[\"item\"]][x[\"node\"]] = d[x[\"item\"]].get(x[\"node\"], 0) + abs(x[\"qty\"])\n",
        "\n",
        "    def merge(self, other):\n",
        "        for it in ITEM_IDS:\n",
        "            for src, dst in ((other.P, self.P), (other.N, self.N)):\n",
        "                for n, v in src[it].items():\n",
        "                    dst[it][n] = max(dst[it].get(n, 0), v)\n",
        "\n",
        "    def stock(self, base=500):\n",
        "        return {i: base + sum(self.P[i].values()) - sum(self.N[i].values()) for i in ITEM_IDS}\n",
        "\n",
        "    def state(self):\n",
        "        return (tuple(tuple(sorted(self.P[i].items())) for i in ITEM_IDS), tuple(tuple(sorted(self.N[i].items())) for i in ITEM_IDS))\n",
        "\n",
        "\n",
        "TX = make_txns(7, 7, per_node_day=60)\n",
        "truth = stock_of(TX)\n",
        "reps = {n: Replica(n) for n in NODES}\n",
        "for x in TX:\n",
        "    reps[x[\"node\"]].apply(x)                                       # 7일간 각 노드는 자기 기록만 반영(오프라인)\n",
        "server = Replica(\"서버\")\n",
        "\n",
        "# 1) 통신 복구: 노드 -> 서버 병합 후 서버 -> 노드 병합(2단계). 지연 측정\n",
        "lat = []\n",
        "t0 = time.perf_counter()\n",
        "for n in NODES:\n",
        "    t1 = time.perf_counter(); server.merge(reps[n]); lat.append((time.perf_counter() - t1) * 1000)\n",
        "for n in NODES:\n",
        "    t1 = time.perf_counter(); reps[n].merge(server); lat.append((time.perf_counter() - t1) * 1000)\n",
        "dt = time.perf_counter() - t0\n",
        "conv = len({r.state() for r in reps.values()} | {server.state()}) == 1\n",
        "lost = sum(abs(server.stock()[i] - truth[i]) for i in ITEM_IDS)\n",
        "print(f\"[CRDT 동기화] 노드 5곳 + 서버, 트랜잭션 {len(TX)}건: 수렴 {'성공' if conv else '실패'}, 서버 재고와 정답(전체 합산) 차이 합계 {lost}개 -> 갱신 유실 {'0건' if lost == 0 else '있음'}\")\n",
        "print(f\"병합 1회 평균 {np.mean(lat):.3f}ms, 최대 {max(lat):.3f}ms, 전체 동기화 {dt * 1000:.2f}ms (목표 200ms 이하: {'달성' if max(lat) <= 200 else '미달'})\")\n",
        "\n",
        "# 2) 비교: 재고 숫자를 '마지막 기록이 이김(LWW)'으로 덮어쓰면 오프라인 갱신이 사라진다\n",
        "last = {}\n",
        "for n in NODES:                                                    # 각 노드의 최종 재고 값(기초 재고 + 자기 기록만 반영)\n",
        "    loc = {i: 500 for i in ITEM_IDS}\n",
        "    for x in TX:\n",
        "        if x[\"node\"] == n:\n",
        "            loc[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    last[n] = (max(x[\"t\"] for x in TX if x[\"node\"] == n), loc)\n",
        "lww = max(last.values())[1]\n",
        "print(f\"숫자 덮어쓰기(LWW) 병합: 정답과 차이 합계 {sum(abs(lww[i] - truth[i]) for i in ITEM_IDS)}개 (갱신 유실) / PN-카운터 {lost}개\")\n",
        "\n",
        "# 3) CRDT 성질 시험: 멱등·교환·결합·중복·오래된 상태 재전송\n",
        "rng = np.random.default_rng(5); ok = {\"멱등(두 번 병합)\": 0, \"교환(A+B = B+A)\": 0, \"결합((A+B)+C = A+(B+C))\": 0, \"오래된 상태 재전송\": 0}\n",
        "trials = 200\n",
        "for _ in range(trials):\n",
        "    a, b, c = (reps[n] for n in rng.choice(NODES, 3, replace=False))\n",
        "    parts = []\n",
        "    for r in (a, b, c):\n",
        "        q = Replica(\"t\")\n",
        "        for i in ITEM_IDS:\n",
        "            for n in NODES:\n",
        "                if rng.random() < .5:\n",
        "                    q.P[i][n] = int(rng.integers(0, 100)); q.N[i][n] = int(rng.integers(0, 100))\n",
        "        parts.append(q)\n",
        "    x, y, z = parts\n",
        "    m = copy.deepcopy(x); m.merge(y); s1 = m.state(); m.merge(y); ok[\"멱등(두 번 병합)\"] += s1 == m.state()\n",
        "    m1 = copy.deepcopy(x); m1.merge(y); m2 = copy.deepcopy(y); m2.merge(x); ok[\"교환(A+B = B+A)\"] += m1.state() == m2.state()\n",
        "    l = copy.deepcopy(x); l.merge(y); l.merge(z); rr = copy.deepcopy(y); rr.merge(z); r2 = copy.deepcopy(x); r2.merge(rr); ok[\"결합((A+B)+C = A+(B+C))\"] += l.state() == r2.state()\n",
        "    new = copy.deepcopy(x); new.merge(y); old = copy.deepcopy(x); before = new.state(); new.merge(old); ok[\"오래된 상태 재전송\"] += before == new.state()\n",
        "print(\"\\n[CRDT 성질 시험(무작위 상태 200쌍)]\", \", \".join(f\"{k} {v}/{trials}\" for k, v in ok.items()))\n",
        "\n",
        "# 4) 메타 정보(최소 보유량·위치·상태) 동시 수정 충돌 탐지: 시간 비교 규칙 vs 버전 벡터 규칙\n",
        "def dominated(a, b):\n",
        "    return all(a.get(k, 0) <= b.get(k, 0) for k in set(a) | set(b)) and a != b\n",
        "\n",
        "\n",
        "def make_pair(kind):\n",
        "    base = float(rng.uniform(0, 160)); va, vb = int(rng.integers(30, 80)), int(rng.integers(30, 80))\n",
        "    if va == vb:\n",
        "        vb += 7\n",
        "    A = {\"field\": \"min_level\", \"value\": va, \"t\": base, \"vv\": {\"A\": 3, \"B\": 1}}\n",
        "    if kind == \"concurrent\":\n",
        "        B = {\"field\": \"min_level\", \"value\": vb, \"t\": base + float(rng.uniform(0.5, 120)), \"vv\": {\"A\": 2, \"B\": 4}}\n",
        "    elif kind == \"causal\":                                      # B가 A의 수정을 받은 뒤 고침(정상 후속 수정), 시각은 가까움\n",
        "        B = {\"field\": \"min_level\", \"value\": vb, \"t\": base + float(rng.uniform(0.01, 0.5)), \"vv\": {\"A\": 3, \"B\": 2}}\n",
        "    elif kind == \"same_value\":\n",
        "        B = {\"field\": \"min_level\", \"value\": va, \"t\": base + float(rng.uniform(0.5, 120)), \"vv\": {\"A\": 2, \"B\": 4}}\n",
        "    elif kind == \"other_field\":\n",
        "        B = {\"field\": \"location\", \"value\": vb, \"t\": base + float(rng.uniform(0.5, 120)), \"vv\": {\"A\": 2, \"B\": 4}}\n",
        "    else:                                                        # 시계 어긋남: 실제로는 동시 수정이지만 B 단말 시계가 느려 시각이 A보다 앞선 것처럼 기록됨\n",
        "        B = {\"field\": \"min_level\", \"value\": vb, \"t\": base - float(rng.uniform(0.5, 3)), \"vv\": {\"A\": 2, \"B\": 4}}\n",
        "    return A, B\n",
        "\n",
        "\n",
        "truth_conf = {\"concurrent\": True, \"causal\": False, \"same_value\": False, \"other_field\": False, \"skew\": True}\n",
        "pairs = [(k, *make_pair(k)) for k in rng.choice(list(truth_conf), 400, p=[.40, .25, .10, .10, .15])]\n",
        "rule_time = lambda A, B: A[\"field\"] == B[\"field\"] and A[\"value\"] != B[\"value\"] and abs(A[\"t\"] - B[\"t\"]) <= 1.0          # 1시간 이내 같은 필드 다른 값\n",
        "rule_val = lambda A, B: A[\"field\"] == B[\"field\"] and A[\"value\"] != B[\"value\"]                                  # 같은 필드 다른 값이면 모두 충돌(인과 무시)\n",
        "rule_vv1 = lambda A, B: A[\"field\"] == B[\"field\"] and not dominated(A[\"vv\"], B[\"vv\"]) and not dominated(B[\"vv\"], A[\"vv\"])      # 동시이면 충돌(값 비교 없음)\n",
        "rule_vv2 = lambda A, B: rule_vv1(A, B) and A[\"value\"] != B[\"value\"]                                                               # 재설계: 값이 다를 때만\n",
        "print(\"\\n[충돌 탐지] 정답 충돌 %d건 / 쌍 %d건\" % (sum(truth_conf[k] for k, _, _ in pairs), len(pairs)))\n",
        "for name, rule in ((\"시간 근접 규칙(1시간 이내)\", rule_time), (\"값 비교 규칙(다르면 충돌)\", rule_val), (\"버전 벡터 규칙 v1(동시이면 충돌)\", rule_vv1), (\"버전 벡터 + 값 비교 v2(재설계)\", rule_vv2)):\n",
        "    pred = np.array([rule(a, b) for _, a, b in pairs]); tr = np.array([truth_conf[k] for k, _, _ in pairs])\n",
        "    tp, fp, fn = int((pred & tr).sum()), int((pred & ~tr).sum()), int((~pred & tr).sum())\n",
        "    kinds = {str(k): int(v) for k, v in pd.Series([k for (k, _, _), p, t in zip(pairs, pred, tr) if p != t]).value_counts().items()}\n",
        "    print(f\"  {name:30s} 재현율 {tp / (tp + fn):5.1%}, 정밀도 {tp / max(tp + fp, 1):5.1%}, 오경보 {fp}건, 놓침 {fn}건 {kinds}\")\n",
        "print(\"오경보·미탐 패턴: 시간 규칙은 오프라인 동안 멀리 떨어진 동시 수정(concurrent)·시계 어긋남(skew)을 놓치고, 값 비교 규칙은 '후속 정상 수정'(causal)을 충돌로 올림 -> 버전 벡터로 인과 관계를 보고 같은 값이면 제외(v2)\")\n",
        "print(\"※ 충돌로 표시된 건은 자동 병합하지 않고 군수관이 수동 병합한다(M7). 재고 수량은 CRDT로 자동 합산되므로 사람이 판단할 대상은 메타 정보 충돌이다.\")\n",
        "\n",
        "plt.figure(figsize=(6, 3.5)); x = np.arange(len(ITEM_IDS))\n",
        "plt.bar(x - .2, [truth[i] for i in ITEM_IDS], .4, label=\"truth / CRDT\"); plt.bar(x + .2, [lww[i] for i in ITEM_IDS], .4, label=\"LWW overwrite\", color=\"tab:red\")\n",
        "plt.xticks(x, [f\"{i + 1}\" for i in range(12)]); plt.xlabel(\"item\"); plt.ylabel(\"stock\"); plt.legend(); plt.title(\"Merged stock: CRDT vs last-write-wins\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[CRDT 동기화] 노드 5곳 + 서버, 트랜잭션 2100건: 수렴 성공, 서버 재고와 정답(전체 합산) 차이 합계 0개 -> 갱신 유실 0건\n",
        "병합 1회 평균 0.022ms, 최대 0.035ms, 전체 동기화 0.22ms (목표 200ms 이하: 달성)\n",
        "숫자 덮어쓰기(LWW) 병합: 정답과 차이 합계 5142개 (갱신 유실) / PN-카운터 0개\n",
        "\n",
        "[CRDT 성질 시험(무작위 상태 200쌍)] 멱등(두 번 병합) 200/200, 교환(A+B = B+A) 200/200, 결합((A+B)+C = A+(B+C)) 200/200, 오래된 상태 재전송 200/200\n",
        "\n",
        "[충돌 탐지] 정답 충돌 226건 / 쌍 400건\n",
        "  시간 근접 규칙(1시간 이내)               재현율  4.9%, 정밀도  9.3%, 오경보 107건, 놓침 215건 {'concurrent': 163, 'causal': 107, 'skew': 52}\n",
        "  값 비교 규칙(다르면 충돌)                재현율 100.0%, 정밀도 67.9%, 오경보 107건, 놓침 0건 {'causal': 107}\n",
        "  버전 벡터 규칙 v1(동시이면 충돌)           재현율 100.0%, 정밀도 88.3%, 오경보 30건, 놓침 0건 {'same_value': 30}\n",
        "  버전 벡터 + 값 비교 v2(재설계)           재현율 100.0%, 정밀도 100.0%, 오경보 0건, 놓침 0건 {}\n",
        "오경보·미탐 패턴: 시간 규칙은 오프라인 동안 멀리 떨어진 동시 수정(concurrent)·시계 어긋남(skew)을 놓치고, 값 비교 규칙은 '후속 정상 수정'(causal)을 충돌로 올림 -> 버전 벡터로 인과 관계를 보고 같은 값이면 제외(v2)\n",
        "※ 충돌로 표시된 건은 자동 병합하지 않고 군수관이 수동 병합한다(M7). 재고 수량은 CRDT로 자동 합산되므로 사람이 판단할 대상은 메타 정보 충돌이다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 05. 오프라인 위변조 방지 암호화 해시 및 데이터 무결성 검증 엔진 개발\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "오프라인 위변조 방지 암호화 해시 및 데이터 무결성 검증 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **보급소 로그의 해시 체인과 레코드별 HMAC 서명으로 위변조를 찾는 오프라인 무결성 검증 엔진을 만들고, 단순 전체 해시 비교 대비 오경보율을 40% 이상 줄이고 신뢰도 등급 산출의 일관성 90% 이상을 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **수량 수정·기록 삭제·순서 변경·위조 삽입·재전송 변조(같은 ID, 다른 수량)·꼬리 절단 6종의 악성 변형과 중복 전송·순서 뒤바뀜·공백 차이 3종의 정상 변형을 보급소 5곳에 6회씩 주입하여 처리 지연 및 복원력을 시험하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/verification/dat10_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": [
        "\"\"\"보급소 로그의 해시 체인·레코드별 HMAC 서명으로 위변조(수량 수정·삭제·순서 변경·위조 삽입·재전송 변조·꼬리 절단)를 탐지하는 오프라인 무결성 검증 엔진과, 점검 항목 가중합 신뢰도(High/Med/Low)를 구현해 단순 전체 해시 비교 대비 오경보율 감축·신뢰도 산출 일관성·불확실성 표시를 검증한다.\"\"\"\n",
        "import hashlib\n",
        "import hmac\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "KEYS = {n: hashlib.sha256(f\"교육용-가상-키-{n}\".encode()).digest() for n in NODES}          # 노드별 서명 키(가상, 실제 키 관리는 별도 체계)\n",
        "TX = make_txns(7, 7, per_node_day=60)\n",
        "\n",
        "\n",
        "def rec_str(x):\n",
        "    return f\"{x['id']}|{x['node']}|{x['item']}|{x['op']}|{x['qty']}|{x['t']}\"\n",
        "\n",
        "\n",
        "def sign_log(node):\n",
        "    \"\"\"노드 로그를 순서대로 해시 체인으로 묶고 레코드별 HMAC 서명을 붙인다. 마지막 해시(머리)는 동료 노드가 따로 보관한다.\"\"\"\n",
        "    log, prev = [], \"GENESIS\"\n",
        "    for x in [y for y in TX if y[\"node\"] == node]:\n",
        "        h = hashlib.sha256((prev + rec_str(x)).encode()).hexdigest()\n",
        "        log.append({**x, \"prev\": prev, \"hash\": h, \"sig\": hmac.new(KEYS[node], rec_str(x).encode(), \"sha256\").hexdigest()[:16], \"note\": \"\"}); prev = h\n",
        "    return log, prev\n",
        "\n",
        "\n",
        "LOGS = {n: sign_log(n) for n in NODES}\n",
        "HEADS = {n: LOGS[n][1] for n in NODES}\n",
        "\n",
        "\n",
        "def canonical(log):\n",
        "    \"\"\"배달 과정의 정상 변형(중복 전송·순서 뒤바뀜·공백 차이)을 정리: 같은 ID는 내용이 같으면 하나만, 일련번호 순 정렬, note 공백 제거.\"\"\"\n",
        "    seen, out, conflicts = {}, [], 0\n",
        "    for r in log:\n",
        "        if r[\"id\"] in seen:\n",
        "            conflicts += (rec_str(r) != rec_str(seen[r[\"id\"]])); continue\n",
        "        seen[r[\"id\"]] = r; out.append(r)\n",
        "    return sorted(out, key=lambda r: r[\"id\"]), conflicts\n",
        "\n",
        "\n",
        "def verify(node, log, head=None):\n",
        "    \"\"\"검증 항목 6개의 결과 dict: 체인 연속, 서명, 같은 ID 내용 충돌, 일련번호 연속, 시각 이상, 꼬리 절단.\"\"\"\n",
        "    log, conf = canonical(log)\n",
        "    chain_ok = sig_ok = True; prev = \"GENESIS\"; bad_sig = 0\n",
        "    for r in log:\n",
        "        if r[\"prev\"] != prev or hashlib.sha256((prev + rec_str(r)).encode()).hexdigest() != r[\"hash\"]:\n",
        "            chain_ok = False\n",
        "        if not hmac.compare_digest(r[\"sig\"], hmac.new(KEYS[node], rec_str(r).encode(), \"sha256\").hexdigest()[:16]):\n",
        "            sig_ok = False; bad_sig += 1\n",
        "        prev = r[\"hash\"]\n",
        "    seqs = [int(r[\"id\"].split(\"-\")[1]) for r in log]\n",
        "    gap = seqs != list(range(1, len(seqs) + 1)) if seqs else True\n",
        "    ts = np.array([r[\"t\"] for r in log]); skew = bool(len(ts) and (ts < 0).any())\n",
        "    tail = bool(head is not None and prev != head)\n",
        "    return {\"chain\": chain_ok, \"sig\": sig_ok, \"conflict\": conf == 0, \"seq\": not gap, \"time\": not skew, \"tail\": not tail, \"bad_sig\": bad_sig}\n",
        "\n",
        "\n",
        "W = {\"chain\": 0.30, \"sig\": 0.30, \"conflict\": 0.15, \"seq\": 0.10, \"time\": 0.05, \"tail\": 0.10}        # 신뢰도 가중치 매트릭스(합 1)\n",
        "\n",
        "\n",
        "def confidence(v, sample_frac=1.0, rng=None):\n",
        "    \"\"\"점검 통과 가중합. 체인·서명·충돌·꼬리 실패는 즉시 Low. 서명은 sample_frac 비율의 표본 점검만 했을 때 불확실성이 생긴다.\"\"\"\n",
        "    s = sum(W[k] for k in W if v[k])\n",
        "    lvl = \"Low\" if not (v[\"chain\"] and v[\"sig\"] and v[\"conflict\"] and v[\"tail\"]) else \"High\" if s >= 0.95 else \"Med\" if s >= 0.75 else \"Low\"\n",
        "    return s, lvl\n",
        "\n",
        "\n",
        "# 시험 1: 악성 변조 6종 + 정상 전송 변형 3종, 각 노드 x 6회\n",
        "import copy\n",
        "rng = np.random.default_rng(11)\n",
        "\n",
        "\n",
        "def tamper(kind, log):\n",
        "    log = copy.deepcopy(log); i = int(rng.integers(5, len(log) - 5))\n",
        "    if kind == \"수량 수정\":\n",
        "        log[i][\"qty\"] += 40\n",
        "    elif kind == \"기록 삭제\":\n",
        "        del log[i]\n",
        "    elif kind == \"순서 바꿈+시각 수정\":\n",
        "        log[i], log[i + 1] = log[i + 1], log[i]; log[i][\"t\"] = round(log[i][\"t\"] + 3, 2)\n",
        "    elif kind == \"위조 삽입(서명 없음)\":\n",
        "        f = dict(log[i]); f[\"id\"] = log[i][\"id\"].split(\"-\")[0] + \"-\" + f\"{len(log) + 1:05d}\"; f[\"qty\"] = 99; f[\"sig\"] = \"0\" * 16; log.append(f)\n",
        "    elif kind == \"재전송 변조(같은 ID, 다른 수량)\":\n",
        "        f = dict(log[i]); f[\"qty\"] += 7; log.append(f)\n",
        "    elif kind == \"꼬리 절단\":\n",
        "        log = log[: -int(rng.integers(3, 20))]\n",
        "    elif kind == \"중복 전송\":\n",
        "        log += copy.deepcopy(log[i:i + 5])\n",
        "    elif kind == \"순서 뒤바뀜 전송\":\n",
        "        rng.shuffle(log)\n",
        "    elif kind == \"공백 차이(note)\":\n",
        "        for r in log[i:i + 3]:\n",
        "            r[\"note\"] = \" \"\n",
        "    return log\n",
        "\n",
        "\n",
        "MAL = [\"수량 수정\", \"기록 삭제\", \"순서 바꿈+시각 수정\", \"위조 삽입(서명 없음)\", \"재전송 변조(같은 ID, 다른 수량)\", \"꼬리 절단\"]\n",
        "BEN = [\"중복 전송\", \"순서 뒤바뀜 전송\", \"공백 차이(note)\"]\n",
        "rows = []\n",
        "for n in NODES:\n",
        "    for kind in MAL + BEN:\n",
        "        for _ in range(6):\n",
        "            lg = tamper(kind, LOGS[n][0])\n",
        "            naive = hashlib.sha256(\"\".join(rec_str(r) + r[\"hash\"] + r[\"note\"] for r in lg).encode()).hexdigest() != hashlib.sha256(\"\".join(rec_str(r) + r[\"hash\"] + r[\"note\"] for r in LOGS[n][0]).encode()).hexdigest()\n",
        "            v = verify(n, lg, HEADS[n]); s, lvl = confidence(v)\n",
        "            rows.append({\"node\": n, \"kind\": kind, \"mal\": kind in MAL, \"naive\": naive, \"alarm\": lvl == \"Low\", \"score\": s, \"lvl\": lvl})\n",
        "R = pd.DataFrame(rows)\n",
        "det = R[R[\"mal\"]].groupby(\"kind\").agg(naive=(\"naive\", \"mean\"), layered=(\"alarm\", \"mean\"))\n",
        "print(f\"[변조 탐지] 악성 {len(MAL)}종 x 노드 5 x 6회 = {int(R['mal'].sum())}건, 정상 전송 변형 {len(BEN)}종 = {int((~R['mal']).sum())}건\")\n",
        "print(\"악성 종류별 탐지율: 전체 해시 비교 / 계층 검증(체인+서명+충돌+순번+꼬리)\")\n",
        "for k, r in det.iterrows():\n",
        "    print(f\"  {k:28s} {r['naive']:5.0%} / {r['layered']:5.0%}\")\n",
        "tp_l, fp_l = int((R[\"alarm\"] & R[\"mal\"]).sum()), int((R[\"alarm\"] & ~R[\"mal\"]).sum())\n",
        "tp_n, fp_n = int((R[\"naive\"] & R[\"mal\"]).sum()), int((R[\"naive\"] & ~R[\"mal\"]).sum())\n",
        "far_n, far_l = fp_n / (tp_n + fp_n), fp_l / max(tp_l + fp_l, 1)\n",
        "print(f\"\\n[오경보] 전체 해시 비교: 경보 {tp_n + fp_n}건 중 정상 변형 {fp_n}건(오경보율 {far_n:.1%}), 탐지율 {tp_n / R['mal'].sum():.0%}\")\n",
        "print(f\"          계층 검증(정규화 후): 경보 {tp_l + fp_l}건 중 정상 변형 {fp_l}건(오경보율 {far_l:.1%}), 탐지율 {tp_l / R['mal'].sum():.0%}\")\n",
        "print(f\"-> 오경보율 {far_n:.1%} -> {far_l:.1%} ({(far_n - far_l) / far_n:.0%} 감축, 목표 40% 이상: {'달성' if (far_n - far_l) / far_n >= 0.4 else '미달'})\")\n",
        "miss = R[R[\"mal\"] & ~R[\"alarm\"]][\"kind\"].value_counts().to_dict()\n",
        "print(f\"계층 검증 미탐: {miss or '없음'} (꼬리 절단은 동료가 보관한 머리 해시가 있어야 탐지)\")\n",
        "\n",
        "# 시험 2: 신뢰도 점수와 일관성(같은 로그를 서명 표본 점검 비율을 바꿔가며 20회 점검했을 때 등급이 같은가)\n",
        "print(\"\\n[신뢰도 가중치 매트릭스]\", W, \"/ High 0.95 이상, Med 0.75 이상, 체인·서명·충돌·꼬리 실패는 즉시 Low\")\n",
        "cons = []\n",
        "for n in NODES:\n",
        "    grades, scores = [], []\n",
        "    for k in range(20):\n",
        "        lg = LOGS[n][0] if k % 2 == 0 else tamper(\"중복 전송\", LOGS[n][0]) if k % 4 == 1 else tamper(\"순서 뒤바뀜 전송\", LOGS[n][0])\n",
        "        s, lvl = confidence(verify(n, lg, HEADS[n])); grades.append(lvl); scores.append(s)\n",
        "    cons.append((n, pd.Series(grades).value_counts().iloc[0] / 20, np.mean(scores), np.std(scores)))\n",
        "cm = np.mean([c[1] for c in cons])\n",
        "print(\"정상 로그 20회 반복(정상 전송 변형 포함) 등급 일관성: \" + \", \".join(f\"{n[-1]} {c:.0%}\" for n, c, _, _ in cons) + f\" -> 평균 {cm:.0%} (목표 90% 이상: {'달성' if cm >= .9 else '미달'})\")\n",
        "lv = pd.crosstab(R[\"kind\"], R[\"lvl\"]).reindex(columns=[\"High\", \"Med\", \"Low\"], fill_value=0)\n",
        "print(\"\\n[변형 종류별 신뢰도 등급 분포]\"); print(lv.to_string())\n",
        "bad = verify(\"보급소A\", tamper(\"수량 수정\", LOGS[\"보급소A\"][0]), HEADS[\"보급소A\"])\n",
        "print(f\"불확실성 표시 예: 수량 수정 로그의 서명 불일치 레코드 {bad['bad_sig']}건 -> 해당 레코드 번호를 군수관에게 제시(전체 폐기 아님)\")\n",
        "print(\"※ 서명 키 배포·보관, 단말 분실 시 폐기는 별도 보안 체계가 필요하다. 신뢰도 등급은 점검 항목 가중합이며 통계적 확률이 아니다. 최종 판단은 군수관·보안 담당이 한다.\")\n",
        "\n",
        "plt.figure(figsize=(8, 3.8)); x = np.arange(len(det))\n",
        "plt.bar(x - .2, det[\"naive\"] * 100, .4, label=\"whole-hash compare\", color=\"tab:gray\"); plt.bar(x + .2, det[\"layered\"] * 100, .4, label=\"layered verify\", color=\"tab:blue\")\n",
        "plt.xticks(x, [f\"M{i + 1}\" for i in range(len(det))]); plt.ylabel(\"detection rate (%)\"); plt.title(\"Tamper detection by type (M1-M6)\"); plt.legend(); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[변조 탐지] 악성 6종 x 노드 5 x 6회 = 180건, 정상 전송 변형 3종 = 90건\n",
        "악성 종류별 탐지율: 전체 해시 비교 / 계층 검증(체인+서명+충돌+순번+꼬리)\n",
        "  기록 삭제                         100% /  100%\n",
        "  꼬리 절단                         100% /  100%\n",
        "  수량 수정                         100% /  100%\n",
        "  순서 바꿈+시각 수정                   100% /  100%\n",
        "  위조 삽입(서명 없음)                  100% /  100%\n",
        "  재전송 변조(같은 ID, 다른 수량)          100% /  100%\n",
        "\n",
        "[오경보] 전체 해시 비교: 경보 270건 중 정상 변형 90건(오경보율 33.3%), 탐지율 100%\n",
        "          계층 검증(정규화 후): 경보 180건 중 정상 변형 0건(오경보율 0.0%), 탐지율 100%\n",
        "-> 오경보율 33.3% -> 0.0% (100% 감축, 목표 40% 이상: 달성)\n",
        "계층 검증 미탐: 없음 (꼬리 절단은 동료가 보관한 머리 해시가 있어야 탐지)\n",
        "\n",
        "[신뢰도 가중치 매트릭스] {'chain': 0.3, 'sig': 0.3, 'conflict': 0.15, 'seq': 0.1, 'time': 0.05, 'tail': 0.1} / High 0.95 이상, Med 0.75 이상, 체인·서명·충돌·꼬리 실패는 즉시 Low\n",
        "정상 로그 20회 반복(정상 전송 변형 포함) 등급 일관성: A 100%, B 100%, C 100%, D 100%, E 100% -> 평균 100% (목표 90% 이상: 달성)\n",
        "\n",
        "[변형 종류별 신뢰도 등급 분포]\n",
        "lvl                   High  Med  Low\n",
        "kind                                \n",
        "공백 차이(note)             30    0    0\n",
        "기록 삭제                    0    0   30\n",
        "꼬리 절단                    0    0   30\n",
        "수량 수정                    0    0   30\n",
        "순서 뒤바뀜 전송               30    0    0\n",
        "순서 바꿈+시각 수정              0    0   30\n",
        "위조 삽입(서명 없음)             0    0   30\n",
        "재전송 변조(같은 ID, 다른 수량)     0    0   30\n",
        "중복 전송                   30    0    0\n",
        "불확실성 표시 예: 수량 수정 로그의 서명 불일치 레코드 1건 -> 해당 레코드 번호를 군수관에게 제시(전체 폐기 아님)\n",
        "※ 서명 키 배포·보관, 단말 분실 시 폐기는 별도 보안 체계가 필요하다. 신뢰도 등급은 점검 항목 가중합이며 통계적 확률이 아니다. 최종 판단은 군수관·보안 담당이 한다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 06. 오프라인/온라인 전환 상태 및 재고 변동 내역 실시간 대시보드 UI 구현\n",
        "**소요시간**: 약 2.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "오프라인/온라인 상태 및 재고 변동 내역 실시간 대시보드 UI를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **보급소별 오프라인/온라인 상태와 품목별 재고 변동 내역, 공급 경보를 보여 주는 실시간 대시보드 화면 모델을 구현하고 갱신 1초 이내·조작 응답 100ms 이내를 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **7일치 트랜잭션 2,100건을 1시간 단위 168회 데이터 스트림으로 흘려 보급소 상태판(온라인·오프라인·미동기 건수·지연 시간), 품목별 재고 게이지(소진 예상 일수와 납기 비교), 변동 내역, 경보 목록 4개 패널의 화면 모델을 갱신 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/ui/dat10_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": [
        "\"\"\"5개 보급소의 오프라인/온라인 전환 상태와 품목별 재고 변동 내역을 시간 단위 모의 데이터 스트림으로 갱신하는 실시간 대시보드 화면 모델(상태판·재고 게이지·변동 내역·경보 패널 4종)과 야간 모드·긴급 경보 강조·조작 응답 시간을 구현해 갱신 주기 1초 이내·UI 응답 100ms 이내를 측정한다.\"\"\"\n",
        "import time\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "THEMES = {\"주간\": {\"bg\": (255, 255, 255), \"fg\": (30, 30, 30), \"ok\": (30, 120, 60), \"warn\": (150, 90, 0), \"crit\": (190, 30, 30), \"off\": (90, 90, 90)},\n",
        "          \"야간\": {\"bg\": (18, 22, 28), \"fg\": (225, 230, 235), \"ok\": (90, 210, 130), \"warn\": (245, 190, 70), \"crit\": (255, 110, 110), \"off\": (160, 165, 170)}}\n",
        "\n",
        "\n",
        "def lum(c):\n",
        "    f = [(v / 255 / 12.92) if v / 255 <= 0.03928 else ((v / 255 + 0.055) / 1.055) ** 2.4 for v in c]\n",
        "    return 0.2126 * f[0] + 0.7152 * f[1] + 0.0722 * f[2]\n",
        "\n",
        "\n",
        "def contrast(a, b):\n",
        "    la, lb = sorted([lum(a), lum(b)], reverse=True)\n",
        "    return (la + 0.05) / (lb + 0.05)\n",
        "\n",
        "\n",
        "class Dashboard:\n",
        "    \"\"\"화면 모델: 매 tick(1시간)마다 새 트랜잭션을 반영하고 4개 패널의 표시 값을 갱신한다. 렌더링은 모델과 분리해 응답 시간을 따로 잰다.\"\"\"\n",
        "    def __init__(self, txns, outages, base=500):\n",
        "        self.tx = sorted(txns, key=lambda x: x[\"t\"]); self.p = 0; self.outages = outages\n",
        "        self.central = {i: base for i in ITEM_IDS}; self.local = {n: {i: 0 for i in ITEM_IDS} for n in NODES}      # 서버 반영 재고 / 노드별 미동기 변동\n",
        "        self.pending = {n: 0 for n in NODES}; self.last_sync = {n: 0.0 for n in NODES}; self.feed = []; self.theme = \"주간\"; self.node_filter = None; self.only_alert = False; self.acked = set(); self.out24 = {i: [] for i in ITEM_IDS}\n",
        "\n",
        "    def online(self, n, t):\n",
        "        return not any(a <= t < b for a, b in self.outages.get(n, []))\n",
        "\n",
        "    def step(self, t):\n",
        "        while self.p < len(self.tx) and self.tx[self.p][\"t\"] < t:\n",
        "            x = self.tx[self.p]; self.p += 1; d = x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "            if self.online(x[\"node\"], x[\"t\"]):\n",
        "                self.central[x[\"item\"]] += d; self.last_sync[x[\"node\"]] = x[\"t\"]\n",
        "            else:\n",
        "                self.local[x[\"node\"]][x[\"item\"]] += d; self.pending[x[\"node\"]] += 1\n",
        "            if x[\"op\"] == \"out\":\n",
        "                self.out24[x[\"item\"]].append((x[\"t\"], x[\"qty\"]))\n",
        "            self.feed.append((x[\"t\"], x[\"node\"], x[\"item\"], x[\"op\"], x[\"qty\"], \"반영\" if self.online(x[\"node\"], x[\"t\"]) else \"대기\"))\n",
        "        for n in NODES:                                                  # 복구된 노드는 대기분을 한꺼번에 반영\n",
        "            if self.online(n, t) and self.pending[n]:\n",
        "                for i, v in self.local[n].items():\n",
        "                    self.central[i] += v; self.local[n][i] = 0\n",
        "                self.pending[n] = 0; self.last_sync[n] = t\n",
        "        return self.view(t)\n",
        "\n",
        "    def view(self, t):\n",
        "        status = [{\"node\": n, \"state\": \"온라인\" if self.online(n, t) else \"오프라인\", \"pending\": self.pending[n], \"lag_h\": round(t - self.last_sync[n], 1)} for n in NODES]\n",
        "        gauges = []\n",
        "        for it in ITEMS:\n",
        "            st = self.central[it[\"item\"]] + sum(self.local[n][it[\"item\"]] for n in NODES)\n",
        "            o24 = sum(q for (tt, q) in self.out24[it[\"item\"]] if tt >= t - 24)\n",
        "            cover = st / max(o24, 1) if t >= 24 else 99.0                      # 최근 24시간 불출량 기준 소진 예상 일수\n",
        "            lvl = \"긴급\" if (st < it[\"min_level\"] or cover < 0.5 * it[\"lead_days\"]) else \"주의\" if cover < it[\"lead_days\"] else \"정상\"\n",
        "            gauges.append({\"item\": it[\"item\"], \"stock\": st, \"min\": it[\"min_level\"], \"lvl\": lvl, \"cover\": cover, \"lead\": it[\"lead_days\"], \"acked\": it[\"item\"] in self.acked})\n",
        "        if self.only_alert:\n",
        "            gauges = [g for g in gauges if g[\"lvl\"] != \"정상\"]\n",
        "        feed = [f for f in self.feed[-200:] if not self.node_filter or f[1] == self.node_filter][-6:]\n",
        "        alerts = [g for g in gauges if g[\"lvl\"] == \"긴급\" and not g[\"acked\"]] + [dict(node=s[\"node\"], lvl=\"장기 오프라인\") for s in status if s[\"lag_h\"] >= 48]\n",
        "        return {\"t\": t, \"status\": status, \"gauges\": gauges, \"feed\": feed, \"alerts\": alerts, \"theme\": self.theme}\n",
        "\n",
        "    # ---- 사용자 조작 핸들러 ----\n",
        "    def set_theme(self, name):\n",
        "        self.theme = name; return self.view(self.tx[self.p - 1][\"t\"] if self.p else 0)\n",
        "\n",
        "    def set_node(self, n):\n",
        "        self.node_filter = n; return self.view(self.tx[self.p - 1][\"t\"])\n",
        "\n",
        "    def toggle_alert(self):\n",
        "        self.only_alert = not self.only_alert; return self.view(self.tx[self.p - 1][\"t\"])\n",
        "\n",
        "    def ack(self, item):\n",
        "        self.acked.add(item); return self.view(self.tx[self.p - 1][\"t\"])\n",
        "\n",
        "\n",
        "rng = np.random.default_rng(21)\n",
        "TX = make_txns(7, 7, per_node_day=60)\n",
        "OUT = {\"보급소B\": [(24.0, 96.0)], \"보급소D\": [(60.0, 120.0), (140.0, 150.0)], \"보급소E\": [(10.0, 168.5)]}        # 오프라인 구간(시간)\n",
        "db = Dashboard(TX, OUT)\n",
        "lat, snaps, off_n = [], {}, []\n",
        "for h in range(1, 169):\n",
        "    t0 = time.perf_counter(); v = db.step(float(h)); lat.append((time.perf_counter() - t0) * 1000)\n",
        "    off_n.append(sum(s[\"state\"] == \"오프라인\" for s in v[\"status\"]))\n",
        "    if h in (12, 72, 168):\n",
        "        snaps[h] = v\n",
        "lat = np.array(lat)\n",
        "print(f\"[모의 스트림] {len(TX)}건 트랜잭션, 168회 갱신(1시간 단위), 오프라인 구간 {sum(len(x) for x in OUT.values())}개\")\n",
        "print(f\"화면 모델 갱신 지연: 평균 {lat.mean():.2f}ms, 95% {np.percentile(lat, 95):.2f}ms, 최대 {lat.max():.2f}ms (목표 1초 이내: {'달성' if lat.max() < 1000 else '미달'})\")\n",
        "for h, v in snaps.items():\n",
        "    print(f\"  {h:3d}h 시점: 오프라인 {sum(s['state'] == '오프라인' for s in v['status'])}곳, 미동기 {sum(s['pending'] for s in v['status'])}건, 긴급/주의 품목 {sum(g['lvl'] == '긴급' for g in v['gauges'])}/{sum(g['lvl'] == '주의' for g in v['gauges'])}종, 경보 {len(v['alerts'])}건\")\n",
        "last = snaps[168]\n",
        "print(\"\\n[보급소 상태판 168h]  \" + \" | \".join(f\"{s['node'][-1]} {s['state']} 대기{s['pending']} 지연{s['lag_h']}h\" for s in last[\"status\"]))\n",
        "print(\"[재고 경보 품목] \" + \", \".join(f\"{g['item']}:{g['stock']}/소진{g['cover']:.1f}일/납기{g['lead']}일({g['lvl']})\" for g in [x for x in last[\"gauges\"] if x[\"lvl\"] != \"정상\"][:4]) + \" ...\")\n",
        "print(\"[변동 내역] \" + \"; \".join(f\"{f[1][-1]}-{f[2]} {f[3]} {f[4]} {f[5]}\" for f in last[\"feed\"][-3:]))\n",
        "\n",
        "# ---- 사용자 조작 시험: 100회씩 응답 시간 ----\n",
        "def timeit(fn, k=100):\n",
        "    r = []\n",
        "    for _ in range(k):\n",
        "        t0 = time.perf_counter(); fn(); r.append((time.perf_counter() - t0) * 1000)\n",
        "    return np.array(r)\n",
        "ops = {\"야간 모드 전환\": lambda: (db.set_theme(\"야간\"), db.set_theme(\"주간\")), \"보급소 필터\": lambda: (db.set_node(\"보급소C\"), db.set_node(None)),\n",
        "       \"경보만 보기\": lambda: (db.toggle_alert(), db.toggle_alert()), \"경보 확인(ack)\": lambda: (db.ack(\"I01\"), db.acked.clear())}\n",
        "print(\"\\n[UI 조작 응답 시간 ms] 조작 / 평균 / 95% / 최대\")\n",
        "worst = 0\n",
        "for k, fn in ops.items():\n",
        "    r = timeit(fn); worst = max(worst, r.max()); print(f\"  {k:12s} {r.mean():6.3f} {np.percentile(r, 95):6.3f} {r.max():6.3f}\")\n",
        "print(f\"-> 최대 {worst:.2f}ms (목표 100ms 이내: {'달성' if worst < 100 else '미달'})\")\n",
        "\n",
        "# ---- 야간 모드 가독성(WCAG 대비 4.5 이상)과 긴급 경보 강조 ----\n",
        "print(\"\\n[테마 대비비(글자-배경)]\")\n",
        "ok = True\n",
        "for nm, th in THEMES.items():\n",
        "    cs = {k: contrast(th[k], th[\"bg\"]) for k in (\"fg\", \"ok\", \"warn\", \"crit\", \"off\")}; ok &= all(c >= 4.5 for c in cs.values())\n",
        "    print(f\"  {nm}: \" + \", \".join(f\"{k} {c:.1f}\" for k, c in cs.items()))\n",
        "print(f\"모든 색상 대비 4.5 이상: {'달성' if ok else '미달'}\")\n",
        "\n",
        "# ---- 렌더링(4개 패널) ----\n",
        "th = THEMES[\"야간\"]; col = lambda k: tuple(c / 255 for c in th[k])\n",
        "t0 = time.perf_counter()\n",
        "fig, ax = plt.subplots(2, 2, figsize=(10, 6), facecolor=col(\"bg\"))\n",
        "for a in ax.ravel():\n",
        "    a.set_facecolor(col(\"bg\")); a.tick_params(colors=col(\"fg\")); [s.set_color(col(\"off\")) for s in a.spines.values()]\n",
        "a = ax[0, 0]; a.set_title(\"Node status (pending tx)\", color=col(\"fg\"))\n",
        "a.bar([s[\"node\"][-1] for s in last[\"status\"]], [s[\"pending\"] for s in last[\"status\"]], color=[col(\"ok\") if s[\"state\"] == \"온라인\" else col(\"crit\") for s in last[\"status\"]])\n",
        "a = ax[0, 1]; a.set_title(\"Stock vs minimum level\", color=col(\"fg\")); ys = np.arange(len(last[\"gauges\"]))\n",
        "a.barh(ys, [g[\"stock\"] for g in last[\"gauges\"]], color=[col({\"정상\": \"ok\", \"주의\": \"warn\", \"긴급\": \"crit\"}[g[\"lvl\"]]) for g in last[\"gauges\"]])\n",
        "a.scatter([g[\"min\"] for g in last[\"gauges\"]], ys, color=col(\"fg\"), marker=\"|\", s=80); a.set_yticks(ys); a.set_yticklabels([g[\"item\"] for g in last[\"gauges\"]], fontsize=6)\n",
        "a = ax[1, 0]; a.set_title(\"Offline nodes over time\", color=col(\"fg\")); a.step(range(1, 169), off_n, color=col(\"warn\")); a.set_xlabel(\"hour\", color=col(\"fg\"))\n",
        "a = ax[1, 1]; a.set_title(\"Alert count by stage\", color=col(\"fg\"))\n",
        "a.bar([\"critical\", \"caution\", \"normal\"], [sum(g[\"lvl\"] == k for g in last[\"gauges\"]) for k in (\"긴급\", \"주의\", \"정상\")], color=[col(\"crit\"), col(\"warn\"), col(\"ok\")])\n",
        "fig.tight_layout(); fig.canvas.draw(); rt = (time.perf_counter() - t0) * 1000\n",
        "print(f\"\\n렌더링 4개 패널 1회 그리기 {rt:.0f}ms (갱신 1초 이내: {'달성' if rt + lat.max() < 1000 else '미달'}, 응답성 100ms는 화면 모델 조작 기준)\")\n",
        "print(\"※ 대시보드는 판단을 대신하지 않는다. 경보 확인(ack)과 재고 조치는 군수관이 한다. 화면의 재고는 동기화된 값과 노드별 미동기 변동의 합이다.\")\n",
        "plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[모의 스트림] 2100건 트랜잭션, 168회 갱신(1시간 단위), 오프라인 구간 4개\n",
        "화면 모델 갱신 지연: 평균 0.14ms, 95% 0.20ms, 최대 1.36ms (목표 1초 이내: 달성)\n",
        "   12h 시점: 오프라인 1곳, 미동기 6건, 긴급/주의 품목 0/0종, 경보 0건\n",
        "   72h 시점: 오프라인 3곳, 미동기 314건, 긴급/주의 품목 7/3종, 경보 9건\n",
        "  168h 시점: 오프라인 1곳, 미동기 400건, 긴급/주의 품목 6/3종, 경보 7건\n",
        "\n",
        "[보급소 상태판 168h]  A 온라인 대기0 지연1.6h | B 온라인 대기0 지연0.1h | C 온라인 대기0 지연0.2h | D 온라인 대기0 지연0.2h | E 오프라인 대기400 지연160.4h\n",
        "[재고 경보 품목] I02:679/소진2.0일/납기8일(긴급), I03:524/소진3.8일/납기13일(긴급), I04:997/소진2.5일/납기6일(긴급), I05:1503/소진4.9일/납기11일(긴급) ...\n",
        "[변동 내역] C-I05 out 15 반영; C-I03 in 42 반영; B-I02 in 87 반영\n",
        "\n",
        "[UI 조작 응답 시간 ms] 조작 / 평균 / 95% / 최대\n",
        "  야간 모드 전환      0.423  0.498  1.075\n",
        "  보급소 필터        0.467  0.503  0.627\n",
        "  경보만 보기        0.462  0.497  0.640\n",
        "  경보 확인(ack)    0.121  0.149  0.218\n",
        "-> 최대 1.08ms (목표 100ms 이내: 달성)\n",
        "\n",
        "[테마 대비비(글자-배경)]\n",
        "  주간: fg 16.7, ok 5.5, warn 5.6, crit 6.2, off 6.9\n",
        "  야간: fg 14.4, ok 9.5, warn 10.7, crit 6.7, off 7.3\n",
        "모든 색상 대비 4.5 이상: 달성\n",
        "\n",
        "렌더링 4개 패널 1회 그리기 349ms (갱신 1초 이내: 달성, 응답성 100ms는 화면 모델 조작 기준)\n",
        "※ 대시보드는 판단을 대신하지 않는다. 경보 확인(ack)과 재고 조치는 군수관이 한다. 화면의 재고는 동기화된 값과 노드별 미동기 변동의 합이다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 07. 동기화 충돌 발생 시 군수관 수동 병합(Conflict Resolution) HITL 워크플로우 설계\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "동기화 충돌 발생 시 군수관 수동 병합(Conflict Resolution) HITL 워크플로우를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **동기화 충돌이 발생했을 때 시스템이 값을 자동 확정하지 않고 군수관이 수동 병합하며, 영향이 큰 건은 지휘관이 승인하는 HITL 워크플로우를 구현해 검토 큐 대기시간과 처리 로깅 성공률 100%를 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **충돌 12건(실사 수량·최소 보유 기준을 두 보급소가 다르게 수정)을 시스템 제안 - 보급담당 상신 - 군수관 검토와 수동 병합 - 확정(차이 25% 이상이면 지휘관 승인) 순서로 처리하고, 병합 전후 값과 사유·근거를 기록하며 반려 사유를 필수로 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/hitl/dat10_approval_workflow.py 및 docs/dat10_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",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "PERM = {\"시스템\": {\"propose\"}, \"보급담당\": {\"submit\", \"redraft\", \"view\"}, \"군수관\": {\"review\", \"merge\", \"confirm\", \"reject\", \"view\"}, \"지휘관\": {\"approve\", \"reject\", \"feedback\", \"view\"}, \"감사관\": {\"view\", \"audit\"}}\n",
        "USERS = {\"sys\": \"시스템\", \"kim\": \"보급담당\", \"lee\": \"군수관\", \"park\": \"지휘관\", \"jung\": \"감사관\"}\n",
        "IMPACT_CUT = 0.25                                                       # 두 값의 차이가 큰 쪽 값의 25% 이상이면 지휘관 승인 필요\n",
        "MAX_REDRAFT = 2\n",
        "FB_CATS = {\"우선순위\", \"자원\", \"절차\", \"기타\"}                          # 지휘관 피드백 입력 폼 분류\n",
        "SLA_H = 4.0                                                             # 군수관 검토 대기 허용 시간(h)\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=\"\", value=None, cat=None, cf=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 in (\"delete_log\", \"edit_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",
        "            a, b = cf\n",
        "            self.cases[case] = {\"state\": \"제안\", \"submitter\": None, \"version\": 1, \"a\": a, \"b\": b, \"merged\": None, \"high\": abs(a - b) / max(abs(a), abs(b), 1) >= IMPACT_CUT, \"feedback\": []}\n",
        "            self.audit.add(event=\"PROPOSE\", who=who, case=case, a=a, b=b, 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 == \"feedback\":\n",
        "            if cat not in FB_CATS:\n",
        "                return self._deny(who, case, action, f\"피드백 분류 오류({cat}), 허용 {sorted(FB_CATS)}\")\n",
        "            c[\"feedback\"].append((cat, reason))\n",
        "            self.audit.add(event=\"FEEDBACK\", who=who, role=role, case=case, cat=cat, reason=reason, evidence=evidence); return True, \"지휘관 피드백 기록\"\n",
        "        if action == \"merge\":\n",
        "            if c[\"state\"] != \"군수관검토\":\n",
        "                return self._deny(who, case, action, f\"병합은 군수관 검토 단계에서만 가능(현재 {c['state']})\")\n",
        "            lo, hi = min(c[\"a\"], c[\"b\"]), max(c[\"a\"], c[\"b\"])\n",
        "            if value is None or not (lo - 0.2 * abs(lo) <= value <= hi + 0.2 * abs(hi)):\n",
        "                return self._deny(who, case, action, f\"병합 값 범위 초과(허용 {lo - 0.2 * abs(lo):.0f}~{hi + 0.2 * abs(hi):.0f}, 입력 {value})\")\n",
        "            before, c[\"merged\"] = c[\"merged\"], value\n",
        "            self.audit.add(event=\"MERGE\", who=who, role=role, case=case, before=before, after=value, a=c[\"a\"], b=c[\"b\"], reason=reason, evidence=evidence); return True, f\"병합 값 {before} -> {value}\"\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 = {\"제안\": \"submit\", \"상신\": \"review\", \"군수관검토\": \"approve\" if c[\"high\"] else \"confirm\"}[c[\"state\"]]\n",
        "        if action != need:\n",
        "            return self._deny(who, case, action, f\"단계 위반(현재 {c['state']}{'·영향 큼' if c['high'] else ''}, 필요 행위 {need})\")\n",
        "        if action == \"confirm\" and c[\"merged\"] is None or action == \"approve\" and c[\"merged\"] is None:\n",
        "            return self._deny(who, case, action, \"군수관 병합 값 없음(자동 확정 금지)\")\n",
        "        if action == \"submit\":\n",
        "            c[\"submitter\"] = who\n",
        "        c[\"state\"] = {\"submit\": \"상신\", \"review\": \"군수관검토\", \"confirm\": \"확정\", \"approve\": \"확정\"}[action]\n",
        "        self.audit.add(event=action.upper(), who=who, role=role, case=case, version=c[\"version\"], to=c[\"state\"], merged=c[\"merged\"], reason=reason, evidence=evidence); return True, c[\"state\"]\n",
        "\n",
        "\n",
        "# ---- 충돌 건 12개: 같은 품목의 재고 실사 수량·최소 보유 기준을 두 보급소가 서로 다르게 수정 ----\n",
        "rng = np.random.default_rng(5)\n",
        "cands = []\n",
        "for k in range(12):\n",
        "    it = ITEMS[k]; field = \"실사 수량\" if k % 3 else \"최소 보유 기준\"\n",
        "    base = 500 if field == \"실사 수량\" else it[\"min_level\"]\n",
        "    a = int(base * rng.uniform(0.9, 1.1)); b = int(a * rng.choice([1.05, 1.1, 1.4, 1.8, 0.5]))\n",
        "    cover = rng.uniform(1.5, 14)                                         # 소진 예상 일수 (납기 대비 부족하면 공급망 재난 징후)\n",
        "    risk = 100 * (0.5 * min(abs(a - b) / max(a, b) / 0.5, 1) + 0.5 * max(0, 1 - cover / it[\"lead_days\"]))\n",
        "    cands.append({\"id\": f\"C{k + 1:02d}\", \"item\": it[\"item\"], \"field\": field, \"a\": a, \"b\": b, \"cover\": cover, \"lead\": it[\"lead_days\"], \"risk\": risk})\n",
        "cd = pd.DataFrame(cands)\n",
        "print(f\"[충돌 건] {len(cd)}건, 영향 큼(차이 {IMPACT_CUT:.0%} 이상) {sum(abs(r.a - r.b) / max(r.a, r.b) >= IMPACT_CUT for r in cd.itertuples())}건, 위험도 평균 {cd['risk'].mean():.0f}\")\n",
        "\n",
        "# ---- 큐 시뮬레이션: 군수관 1명(건당 2시간), 건이 1시간 간격으로 도착 ----\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(cd[\"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 >= 50]\n",
        "print(f\"\\n[검토 큐 대기시간(h)] 건 {len(rs)}개, 위험 50 이상 {len(hi)}건, 허용 {SLA_H:.0f}h\")\n",
        "for nm, w in ((\"선입선출\", wf_), (\"위험도 우선(노화 가산)\", wp_)):\n",
        "    print(f\"  {nm:14s}: 전체 평균 {np.mean(list(w.values())):.1f}h, 고위험 평균 {np.mean([w[i] for i in hi]):.1f}h, 최대 {max(w.values()):.1f}h, 허용 초과 {sum(v > SLA_H for v in w.values())}건\")\n",
        "\n",
        "# ---- 워크플로우 모의 ----\n",
        "wf = Workflow(); res = []\n",
        "for k, r in cd.iterrows():\n",
        "    case = r[\"id\"]\n",
        "    wf.act(\"sys\", case, \"propose\", reason=f\"{r['item']} {r['field']} 충돌: A {r['a']} / B {r['b']}\", cf=(int(r[\"a\"]), int(r[\"b\"])))\n",
        "    wf.act(\"kim\", case, \"submit\", \"수동 병합 검토 요청\", f\"소진 예상 {r['cover']:.1f}일, 납기 {r['lead']}일\")\n",
        "    for _ in range(12):\n",
        "        c = wf.cases[case]; s = c[\"state\"]\n",
        "        if s in (\"확정\", \"종결(반려)\"):\n",
        "            break\n",
        "        if s == \"상신\":\n",
        "            if k in (2, 11) and c[\"version\"] <= (1 if k == 2 else 9):           # 2번 건은 1회 반려, 11번 건은 보완 없이 계속 반려(재기안 한도 시연)\n",
        "                wf.act(\"lee\", case, \"reject\", \"근거 사진·실사표 누락, 보완 요청\", f\"충돌 두 값 {r['a']}/{r['b']}\")\n",
        "            else:\n",
        "                wf.act(\"lee\", case, \"review\", \"두 기록과 실사 근거 확인\", f\"소진 예상 {r['cover']:.1f}일\")\n",
        "        elif s == \"군수관검토\":\n",
        "            if c[\"merged\"] is None:\n",
        "                val = int((c[\"a\"] + c[\"b\"]) / 2) if r[\"field\"] == \"실사 수량\" else max(c[\"a\"], c[\"b\"])           # 실사는 평균 후보, 기준은 높은 쪽(안전 우선)\n",
        "                wf.act(\"lee\", case, \"merge\", \"수동 병합: 현장 실사·납기 고려해 값 결정\", f\"A {c['a']} / B {c['b']}\", value=val)\n",
        "            if c[\"high\"]:\n",
        "                wf.act(\"park\", case, \"feedback\", \"영향이 큰 건이므로 선별 확인 후 결정\", \"우선순위\", cat=\"우선순위\")\n",
        "                wf.act(\"park\", case, \"approve\", \"승인: 병합 값 확정\", f\"차이 {abs(c['a'] - c['b']) / max(c['a'], c['b']):.0%}, 군수관 병합 검토 완료\")\n",
        "            else:\n",
        "                wf.act(\"lee\", case, \"confirm\", \"확정: 병합 값 반영\", \"차이가 작아 군수관 확정 범위\")\n",
        "        elif s == \"반려\":\n",
        "            wf.act(\"kim\", case, \"redraft\", \"반려 사유 반영해 재기안\", f\"근거 보완 {c['version']}차\")\n",
        "    c = wf.cases[case]; res.append((case, r[\"field\"], c[\"a\"], c[\"b\"], \"-\" if c[\"merged\"] is None else str(c[\"merged\"]), \"지휘관\" if c[\"high\"] else \"군수관\", c[\"state\"], c[\"version\"]))\n",
        "print(\"\\n[처리 결과] (모의 결재자 기준, 실제 판단은 사람이 한다)\")\n",
        "print(pd.DataFrame(res, columns=[\"건\", \"항목\", \"A값\", \"B값\", \"병합값\", \"확정권자\", \"상태\", \"버전\"]).head(12).to_string(index=False))\n",
        "print(f\"군수관 수동 병합 {sum(e['event'] == 'MERGE' 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\", ab=(100, 160)):\n",
        "    wf.act(\"sys\", cid, \"propose\", reason=\"시험\", cf=ab)\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",
        "done_case = next(c for c, v in wf.cases.items() if v[\"state\"] == \"확정\")\n",
        "att = [(\"시스템 자동 병합\", lambda: wf.act(\"sys\", mk(\"T01\", \"review\"), \"merge\", \"자동\", \"자동\", value=130)), (\"보급담당이 병합 시도\", lambda: wf.act(\"kim\", mk(\"T02\", \"review\"), \"merge\", \"임의\", \"근거\", value=130)),\n",
        "       (\"병합 없이 확정\", lambda: wf.act(\"lee\", mk(\"T03\", \"review\", (100, 105)), \"confirm\", \"바로 확정\", \"근거\")), (\"미등록 사용자\", lambda: wf.act(\"guest\", mk(\"T04\"), \"review\", \"열람\", \"근거\")),\n",
        "       (\"사유 없는 검토\", lambda: wf.act(\"lee\", mk(\"T05\"), \"review\", \"\", \"\")), (\"감사 로그 수정\", lambda: wf.act(\"kim\", mk(\"T06\"), \"edit_log\", \"정리\", \"근거\")),\n",
        "       (\"감사관이 확정 시도\", lambda: wf.act(\"jung\", mk(\"T07\", \"review\"), \"confirm\", \"확정\", \"근거\")), (\"존재하지 않는 건\", lambda: wf.act(\"park\", \"X99\", \"approve\", \"승인\", \"근거\")),\n",
        "       (\"종결된 건 재승인\", lambda: wf.act(\"park\", done_case, \"approve\", \"재승인\", \"근거\")), (\"병합 값 범위 초과(100/160 -> 500)\", lambda: wf.act(\"lee\", mk(\"T08\", \"review\"), \"merge\", \"대폭 변경\", \"근거\", value=500)),\n",
        "       (\"영향 큰 건을 군수관이 확정\", lambda: (wf.act(\"lee\", \"T09\", \"merge\", \"병합\", \"근거\", value=130) if mk(\"T09\", \"review\") else None, wf.act(\"lee\", \"T09\", \"confirm\", \"확정\", \"근거\"))[-1]),\n",
        "       (\"지휘관이 검토 건너뛰기\", lambda: wf.act(\"park\", mk(\"T10\"), \"approve\", \"바로 승인\", \"근거\")), (\"피드백 분류 오류\", lambda: wf.act(\"park\", mk(\"T11\", \"review\"), \"feedback\", \"의견\", \"근거\", cat=\"잡담\")),\n",
        "       (\"상신 전 병합\", lambda: wf.act(\"lee\", mk(\"T12\", \"submit\"), \"merge\", \"병합\", \"근거\", value=130))]\n",
        "print(\"\\n[권한 우회·절차 위반 시도]\")\n",
        "blocked = 0\n",
        "for n, fn in att:\n",
        "    ok, why = fn(); blocked += (not ok); print(f\"  {n:26s}-> {'차단' 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 conflicts (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",
        "[충돌 건] 12건, 영향 큼(차이 25% 이상) 5건, 위험도 평균 37\n",
        "\n",
        "[검토 큐 대기시간(h)] 건 12개, 위험 50 이상 6건, 허용 4h\n",
        "  선입선출          : 전체 평균 5.5h, 고위험 평균 5.5h, 최대 11.0h, 허용 초과 7건\n",
        "  위험도 우선(노화 가산) : 전체 평균 5.5h, 고위험 평균 0.8h, 최대 13.0h, 허용 초과 5건\n",
        "\n",
        "[처리 결과] (모의 결재자 기준, 실제 판단은 사람이 한다)\n",
        "  건       항목  A값  B값 병합값 확정권자     상태  버전\n",
        "C01 최소 보유 기준  31  32  32  군수관     확정   1\n",
        "C02    실사 수량 478 239 358  지휘관     확정   1\n",
        "C03    실사 수량 488 683 585  지휘관     확정   2\n",
        "C04 최소 보유 기준  54  75  75  지휘관     확정   1\n",
        "C05    실사 수량 515 927 721  지휘관     확정   1\n",
        "C06    실사 수량 547 601 574  군수관     확정   1\n",
        "C07 최소 보유 기준  42  44  44  군수관     확정   1\n",
        "C08    실사 수량 517 568 542  군수관     확정   1\n",
        "C09    실사 수량 505 252 378  지휘관     확정   1\n",
        "C10 최소 보유 기준  63  69  69  군수관     확정   1\n",
        "C11    실사 수량 537 590 563  군수관     확정   1\n",
        "C12    실사 수량 537 590   -  군수관 종결(반려)   3\n",
        "군수관 수동 병합 11건(전후 값·사유 기록), 반려 4건(사유 필수), 지휘관 피드백 5건\n",
        "\n",
        "[권한 우회·절차 위반 시도]\n",
        "  시스템 자동 병합                 -> 차단: 권한 없음(시스템 역할에 merge 권한 없음)\n",
        "  보급담당이 병합 시도               -> 차단: 권한 없음(보급담당 역할에 merge 권한 없음)\n",
        "  병합 없이 확정                  -> 차단: 군수관 병합 값 없음(자동 확정 금지)\n",
        "  미등록 사용자                   -> 차단: 미등록 사용자\n",
        "  사유 없는 검토                  -> 차단: 사유·근거 입력 필수(반려·병합 사유 포함)\n",
        "  감사 로그 수정                  -> 차단: 감사 로그 삭제·수정 권한은 누구에게도 없음\n",
        "  감사관이 확정 시도                -> 차단: 권한 없음(감사관 역할에 confirm 권한 없음)\n",
        "  존재하지 않는 건                 -> 차단: 존재하지 않는 건\n",
        "  종결된 건 재승인                 -> 차단: 이미 종결된 건(확정)\n",
        "  병합 값 범위 초과(100/160 -> 500)-> 차단: 병합 값 범위 초과(허용 80~192, 입력 500)\n",
        "  영향 큰 건을 군수관이 확정           -> 차단: 단계 위반(현재 군수관검토·영향 큼, 필요 행위 approve)\n",
        "  지휘관이 검토 건너뛰기              -> 차단: 단계 위반(현재 상신·영향 큼, 필요 행위 review)\n",
        "  피드백 분류 오류                 -> 차단: 피드백 분류 오류(잡담), 허용 ['기타', '우선순위', '자원', '절차']\n",
        "  상신 전 병합                   -> 차단: 병합은 군수관 검토 단계에서만 가능(현재 상신)\n",
        "차단 14/14 (100%)\n",
        "\n",
        "처리 요청 115건 -> 감사 로그 115건(확정·반려·병합·거부 모두 기록), 로깅 성공률 100% (목표 100%: 달성)\n",
        "... (이하 생략)\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 08. 7일간 통신 단절 후 대량 재고 트랜잭션 일괄 동기화 스트레스 테스트\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "7일간 통신 단절 후 대량 재고 트랜잭션 일괄 동기화 스트레스 테스트를 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **7일간 통신이 끊겼다가 복구된 직후 보급소 5곳의 누적 트랜잭션이 한꺼번에 서버로 몰리는 상황을 기준 100%·200%·500%로 시험해 시스템 다운 0건·데이터 유실 0%와 기능 축소 동작을 확인한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **트랜잭션 2,100건을 기준으로 4,200건·10,500건까지 늘려 배치 50건 단위로 큐와 작업자 스레드 4개가 처리하게 하고, 같은 배치 중복 재전송(5%)·서명 후 수량 변조(1%)·작업자 오류(37배치마다) 주입하여 처리 지연 및 복원력을 시험하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `tests/dat10_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": [
        "\"\"\"7일간 통신이 끊긴 보급소 5곳의 누적 트랜잭션을 복구 직후 서버로 한꺼번에 올리는 상황을 기준 100%·200%·500%로 늘려 큐 기반 작업자 스레드로 처리하고, 중복 재전송·서명 불일치·작업자 오류를 주입해 다운 0건·유실 0건·기능 축소(표본 서명 점검 후 사후 검증)·병목(목록 탐색 중복 확인을 집합으로 교체)·메모리 증가를 측정한다.\"\"\"\n",
        "import hashlib\n",
        "import hmac\n",
        "import queue\n",
        "import threading\n",
        "import time\n",
        "import tracemalloc\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "# ---------------------------------------------------------------------------\n",
        "# [교육용가상자료] 가상 야전 보급소·업체·품목 자료 생성기 (실제 부대·업체·품목·재고 정보 없음)\n",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "KEYS = {n: hashlib.sha256(f\"demo-key-{n}\".encode()).digest() for n in NODES}            # [교육용가상자료] 데모 키\n",
        "BATCH, QMAX, HIGH_WATER, WORKERS = 50, 12, 8, 4\n",
        "\n",
        "\n",
        "def sig_of(x):\n",
        "    return hmac.new(KEYS[x[\"node\"]], f\"{x['id']}|{x['item']}|{x['op']}|{x['qty']}|{x['t']}\".encode(), \"sha256\").hexdigest()[:16]\n",
        "\n",
        "\n",
        "class Server:\n",
        "    def __init__(self, fast):\n",
        "        self.fast, self.lock = fast, threading.Lock()\n",
        "        self.ids = set() if fast else []                                # 최적화 전: 목록 선형 탐색으로 중복 확인\n",
        "        self.stock = {i: 500 for i in ITEM_IDS}; self.quar, self.dups, self.unver, self.lite_batches, self.applied = [], 0, [], 0, 0\n",
        "\n",
        "    def ingest(self, batch, lite):\n",
        "        ok_rows, unv = [], set()\n",
        "        for k, x in enumerate(batch):\n",
        "            if lite and k % 10:                                         # 기능 축소: 표본 10%만 서명 점검, 나머지는 사후 검증 대상\n",
        "                good = True; unv.add(x[\"id\"])\n",
        "            else:\n",
        "                good = hmac.compare_digest(sig_of(x), x[\"sig\"])\n",
        "            if not good:\n",
        "                with self.lock:\n",
        "                    self.quar.append(x[\"id\"])\n",
        "                continue\n",
        "            ok_rows.append(x)\n",
        "        with self.lock:\n",
        "            self.lite_batches += lite\n",
        "            for x in ok_rows:\n",
        "                dup = x[\"id\"] in self.ids\n",
        "                if dup:\n",
        "                    self.dups += 1; continue\n",
        "                self.ids.add(x[\"id\"]) if self.fast else self.ids.append(x[\"id\"])\n",
        "                self.stock[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]; self.applied += 1\n",
        "                if x[\"id\"] in unv:\n",
        "                    self.unver.append(x)                                # 실제로 반영된 미검증 레코드만 사후 검증 대상\n",
        "\n",
        "    def post_verify(self):\n",
        "        \"\"\"사후 검증: 표본 점검만 거친 레코드를 전수 점검하고, 위조로 확인되면 반영분을 되돌린다.\"\"\"\n",
        "        rolled = 0\n",
        "        for x in self.unver:\n",
        "            if not hmac.compare_digest(sig_of(x), x[\"sig\"]):\n",
        "                self.stock[x[\"item\"]] -= x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]; self.quar.append(x[\"id\"]); self.applied -= 1; rolled += 1\n",
        "        return rolled\n",
        "\n",
        "\n",
        "def run(mult, fast, seed=3):\n",
        "    rng = np.random.default_rng(seed)\n",
        "    tx = make_txns(7, 7, per_node_day=60 * mult)\n",
        "    for x in tx:\n",
        "        x[\"sig\"] = sig_of(x)\n",
        "    bad = set(rng.choice(len(tx), int(len(tx) * 0.01), replace=False).tolist())\n",
        "    for i in bad:\n",
        "        tx[i][\"qty\"] += 13                                              # 서명 후 수량 변조(1%)\n",
        "    valid = [x for i, x in enumerate(tx) if i not in bad]\n",
        "    truth = stock_of(valid)\n",
        "    srv, q, lat, depth, dead, crash = Server(fast), queue.Queue(maxsize=QMAX), [], [], [], [0]\n",
        "    lk = threading.Lock(); cnt = {\"b\": 0}\n",
        "\n",
        "    def worker():\n",
        "        while True:\n",
        "            item = q.get()\n",
        "            if item is None:\n",
        "                return\n",
        "            batch, t_in, retry = item\n",
        "            depth.append(q.qsize())\n",
        "            try:\n",
        "                with lk:\n",
        "                    cnt[\"b\"] += 1; inj = (cnt[\"b\"] % 37 == 0) and not retry                  # 37번째 배치마다 작업자 오류 주입(재시도 1회)\n",
        "                if inj:\n",
        "                    raise RuntimeError(\"worker failure(injected)\")\n",
        "                srv.ingest(batch, lite=fast and q.qsize() >= HIGH_WATER)\n",
        "            except RuntimeError:\n",
        "                dead.append((batch, t_in))                                                # 데드레터 보관 후 재처리(유실 방지)\n",
        "            except Exception:\n",
        "                crash[0] += 1\n",
        "            lat.append((time.perf_counter() - t_in) * 1000)\n",
        "\n",
        "    ths = [threading.Thread(target=worker) for _ in range(WORKERS)]\n",
        "    [t.start() for t in ths]\n",
        "    t0 = time.perf_counter()\n",
        "    batches = []\n",
        "    for n in NODES:\n",
        "        mine = [x for x in tx if x[\"node\"] == n]\n",
        "        batches += [mine[i:i + BATCH] for i in range(0, len(mine), BATCH)]\n",
        "    order = rng.permutation(len(batches)); resend = set(rng.choice(len(batches), max(1, len(batches) // 20), replace=False).tolist())\n",
        "    for i in order:\n",
        "        q.put((batches[i], time.perf_counter(), False))                                  # 가득 차면 생산자가 기다린다(역압, 유실 없음)\n",
        "        if i in resend:\n",
        "            q.put((batches[i], time.perf_counter(), False))                              # 같은 배치 중복 재전송\n",
        "    for _ in ths:\n",
        "        q.put(None)\n",
        "    [t.join() for t in ths]\n",
        "    for b, t_in in dead:\n",
        "        srv.ingest(b, lite=False)\n",
        "    rolled = srv.post_verify() if fast else 0\n",
        "    dt = time.perf_counter() - t0\n",
        "    diff = sum(abs(srv.stock[i] - truth[i]) for i in ITEM_IDS)\n",
        "    lost = len(valid) - srv.applied\n",
        "    return {\"n\": len(tx), \"dt\": dt, \"tps\": len(tx) / dt, \"p95\": float(np.percentile(lat, 95)), \"qmax\": max(depth) if depth else 0, \"lite\": srv.lite_batches, \"dead\": len(dead),\n",
        "            \"dups\": srv.dups, \"quar\": len(set(srv.quar)), \"bad\": len(bad), \"lost\": lost, \"diff\": diff, \"crash\": crash[0], \"rolled\": rolled}\n",
        "\n",
        "\n",
        "print(f\"[7일 단절 후 일괄 동기화] 기준 {5 * 7 * 60}건(5곳 x 7일 x 60건) -> 100%/200%/500%, 배치 {BATCH}건, 작업자 {WORKERS}, 큐 {QMAX}, 중복 재전송 5%, 위조 1%, 오류 주입 37배치마다\")\n",
        "rows = []\n",
        "for mult in (1, 2, 5):\n",
        "    r0, r1 = run(mult, False), run(mult, True); rows.append((mult, r0, r1))\n",
        "    print(f\"  {mult * 100:3d}% ({r1['n']}건): 최적화 전 {r0['dt']:.2f}초({r0['tps']:.0f}건/초, p95 {r0['p95']:.0f}ms) -> 후 {r1['dt']:.2f}초({r1['tps']:.0f}건/초, p95 {r1['p95']:.0f}ms), 큐 최대 {r1['qmax']}, 축소 배치 {r1['lite']}\")\n",
        "print(\"\\n[500% 상세] 항목: 최적화 전 / 후\")\n",
        "a, b = rows[-1][1], rows[-1][2]\n",
        "for k, lab in ((\"dups\", \"중복 거부\"), (\"quar\", \"위조 격리\"), (\"dead\", \"작업자 오류 재처리 배치\"), (\"rolled\", \"사후 검증 되돌림\"), (\"lost\", \"유실 건수\"), (\"diff\", \"재고 차이 합계\"), (\"crash\", \"미처리 예외(다운)\")):\n",
        "    print(f\"  {lab:16s} {a[k]:6d} / {b[k]:6d}\")\n",
        "lost_all = sum(r[1][\"lost\"] + r[2][\"lost\"] for r in rows); diff_all = sum(r[1][\"diff\"] + r[2][\"diff\"] for r in rows); crash_all = sum(r[1][\"crash\"] + r[2][\"crash\"] for r in rows)\n",
        "quar_ok = all(r[2][\"quar\"] == r[2][\"bad\"] for r in rows)\n",
        "print(f\"\\n전체 6회 실행: 다운 {crash_all}건, 유실 {lost_all}건, 재고 차이 {diff_all}, 위조 전부 격리 {'예' if quar_ok else '아니오'} (목표 다운 0건·유실 0%: {'달성' if crash_all == 0 and lost_all == 0 and diff_all == 0 else '미달'})\")\n",
        "print(f\"병목 최적화(500%): 중복 확인을 목록 탐색 -> 집합 조회로 교체해 {a['dt']:.2f}초 -> {b['dt']:.2f}초({a['dt'] / b['dt']:.1f}배), 처리량 {a['tps']:.0f} -> {b['tps']:.0f}건/초\")\n",
        "\n",
        "# ---- 단계별 병목 프로파일링(수동 타이머): 서명 점검 / 중복 확인(목록·집합) ----\n",
        "tx5 = make_txns(7, 7, per_node_day=300)\n",
        "for x in tx5:\n",
        "    x[\"sig\"] = sig_of(x)\n",
        "t0 = time.perf_counter(); [hmac.compare_digest(sig_of(x), x[\"sig\"]) for x in tx5]; t_sig = time.perf_counter() - t0\n",
        "ids = [x[\"id\"] for x in tx5]; t0 = time.perf_counter(); lst = []\n",
        "for i in ids:\n",
        "    _ = i in lst; lst.append(i)\n",
        "t_lst = time.perf_counter() - t0; t0 = time.perf_counter(); st = set()\n",
        "for i in ids:\n",
        "    _ = i in st; st.add(i)\n",
        "t_set = time.perf_counter() - t0\n",
        "print(f\"\\n[병목 프로파일] {len(ids)}건 기준: 서명 점검 {t_sig * 1000:.0f}ms, 중복 확인 목록 {t_lst * 1000:.0f}ms / 집합 {t_set * 1000:.1f}ms -> 병목은 중복 확인이었고 서명 점검이 다음 병목\")\n",
        "print(\"[기능 축소 정책] 큐가 %d건 이상 쌓이면 서명을 표본 10%%만 점검하고 나머지는 사후 전수 검증한다. 위조가 확인되면 되돌리고 격리한다. 데이터는 버리지 않는다.\" % HIGH_WATER)\n",
        "\n",
        "# ---- 메모리 누수 점검: 같은 부하를 3회 반복해 남는 메모리 ----\n",
        "tracemalloc.start(); mem = []\n",
        "for _ in range(3):\n",
        "    run(1, True); mem.append(tracemalloc.get_traced_memory()[0] / 1e6)\n",
        "tracemalloc.stop()\n",
        "print(f\"[메모리] 반복 실행 후 남은 메모리 {', '.join(f'{m:.1f}MB' for m in mem)} -> 증가 {mem[-1] - mem[0]:+.1f}MB ({'누수 징후 없음' if mem[-1] - mem[0] < 5 else '확인 필요'})\")\n",
        "print(\"※ 처리 시간은 실행 환경에 따라 달라진다. 실제 서버는 별도 작업 큐·영속 저장소·부하 시험 도구가 필요하다. 이 시험은 단일 프로세스 모의이다.\")\n",
        "\n",
        "plt.figure(figsize=(7, 3.6)); x = np.arange(3)\n",
        "plt.bar(x - .2, [r[1][\"dt\"] for r in rows], .4, label=\"before (list scan)\", color=\"tab:gray\"); plt.bar(x + .2, [r[2][\"dt\"] for r in rows], .4, label=\"after (set + lite mode)\", color=\"tab:blue\")\n",
        "plt.xticks(x, [\"100%\", \"200%\", \"500%\"]); plt.ylabel(\"seconds\"); plt.title(\"Bulk sync time after 7-day outage\"); plt.legend(); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[7일 단절 후 일괄 동기화] 기준 2100건(5곳 x 7일 x 60건) -> 100%/200%/500%, 배치 50건, 작업자 4, 큐 12, 중복 재전송 5%, 위조 1%, 오류 주입 37배치마다\n",
        "  100% (2100건): 최적화 전 0.05초(45132건/초, p95 21ms) -> 후 0.01초(161281건/초, p95 2ms), 큐 최대 11, 축소 배치 15\n",
        "  200% (4200건): 최적화 전 0.16초(26448건/초, p95 45ms) -> 후 0.03초(126250건/초, p95 6ms), 큐 최대 11, 축소 배치 27\n",
        "  500% (10500건): 최적화 전 1.00초(10489건/초, p95 134ms) -> 후 0.06초(165580건/초, p95 3ms), 큐 최대 11, 축소 배치 68\n",
        "\n",
        "[500% 상세] 항목: 최적화 전 / 후\n",
        "  중복 거부               496 /    497\n",
        "  위조 격리               105 /    105\n",
        "  작업자 오류 재처리 배치         5 /      5\n",
        "  사후 검증 되돌림             0 /     32\n",
        "  유실 건수                 0 /      0\n",
        "  재고 차이 합계              0 /      0\n",
        "  미처리 예외(다운)            0 /      0\n",
        "\n",
        "전체 6회 실행: 다운 0건, 유실 0건, 재고 차이 0, 위조 전부 격리 예 (목표 다운 0건·유실 0%: 달성)\n",
        "병목 최적화(500%): 중복 확인을 목록 탐색 -> 집합 조회로 교체해 1.00초 -> 0.06초(15.8배), 처리량 10489 -> 165580건/초\n",
        "\n",
        "[병목 프로파일] 10500건 기준: 서명 점검 41ms, 중복 확인 목록 765ms / 집합 2.1ms -> 병목은 중복 확인이었고 서명 점검이 다음 병목\n",
        "[기능 축소 정책] 큐가 8건 이상 쌓이면 서명을 표본 10%만 점검하고 나머지는 사후 전수 검증한다. 위조가 확인되면 되돌리고 격리한다. 데이터는 버리지 않는다.\n",
        "[메모리] 반복 실행 후 남은 메모리 0.0MB, 0.0MB, 0.0MB -> 증가 +0.0MB (누수 징후 없음)\n",
        "※ 처리 시간은 실행 환경에 따라 달라진다. 실제 서버는 별도 작업 큐·영속 저장소·부하 시험 도구가 필요하다. 이 시험은 단일 프로세스 모의이다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 09. AAR 기반 동기화 누락 및 데이터 충돌 사례 분석 및 동기화 룰셋 재튜닝\n",
        "**소요시간**: 약 3.0시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "AAR 기반 동기화 누락 및 데이터 충돌 사례를 분석하고 동기화 룰을 재튜닝한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **동기화 누락·충돌 판정 오류 사례를 AAR로 원인별로 분류하고 규칙 파라미터(중복 판정 방식, 충돌 탐지 규칙, 자동 병합 임계, 재시도 횟수)를 재튜닝해 종합 정확도 10%p 이상 향상과 재발 오류 0건을 시험한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **가상 사례 1,200건(중복 재전송·재입력, 같은 크기의 서로 다른 거래, 동시·후속 메타 수정과 시계 어긋남, 수량 차이, 통신불량 전송 실패)에 현행 규칙을 적용해 오류를 D1~D6 원인으로 분류 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `docs/dat10_aar_report.md 및 configs/dat10_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": [
        "\"\"\"동기화 사후 검토(AAR) 사례 1,200건에서 판정 오류를 원인(중복 판정 방식·시계 어긋남·충돌 탐지 규칙·자동 병합 임계·재시도 횟수)별로 분류하고 규칙 파라미터를 재튜닝해, 별도 시험 3,000건의 종합 정확도 향상폭·AAR 사례 재발 오류·회귀 테스트 통과를 계산하고 튜닝 결과를 YAML 문자열로 출력한다.\"\"\"\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "KINDS = [\"dup_resend\", \"distinct_similar\", \"meta_conflict\", \"meta_causal\", \"qty_diff\", \"send_fail\", \"dup_reinput\"]\n",
        "PK = [.20, .12, .18, .15, .15, .15, .05]\n",
        "CAUSE = {\"dup_resend\": \"D2 재전송 중복 놓침\", \"distinct_similar\": \"D1 다른 거래를 중복으로 판정\", \"meta_conflict\": \"D4 동시 수정 놓침(조용한 덮어쓰기)\", \"meta_causal\": \"D3 후속 수정의 승자 오판·불필요 경보\",\n",
        "         \"qty_diff\": \"D5 자동 병합 임계 부적절\", \"send_fail\": \"D6 재시도 부족 누락\", \"dup_reinput\": \"D2b 재입력(새 ID) 중복 놓침\"}\n",
        "\n",
        "\n",
        "def make_events(seed, n):\n",
        "    \"\"\"[교육용가상자료] 동기화 사례 생성. 단말 시계 오차(+-2시간), 버전 벡터 없는 구형 기록 8%, 통신불량 상황의 전송 실패를 포함한다.\"\"\"\n",
        "    rng = np.random.default_rng(seed); out = []\n",
        "    for i in range(n):\n",
        "        k = str(rng.choice(KINDS, p=PK)); o = rng.uniform(-2, 2, 2); ev = {\"id\": i, \"kind\": k, \"vv\": rng.random() > 0.08}\n",
        "        if k == \"dup_resend\":\n",
        "            ev.update(same_id=True, dt=0.0 if rng.random() < .5 else float(rng.uniform(0.5, 6)))\n",
        "        elif k == \"distinct_similar\":\n",
        "            ev.update(same_id=False, dt=float(rng.uniform(0, 0.5)))\n",
        "        elif k == \"dup_reinput\":                                           # 오입력 정정 등으로 같은 내용을 새 ID로 다시 입력\n",
        "            ev.update(same_id=False, dt=float(rng.uniform(0.1, 6)))\n",
        "        elif k in (\"meta_conflict\", \"meta_causal\"):\n",
        "            gap = float(rng.uniform(0.5, 60)) if k == \"meta_conflict\" else float(rng.uniform(0.01, 1.5))\n",
        "            ev.update(ta=100 + o[0], tb=100 + gap + o[1])                  # 기록된 시각 = 실제 시각 + 단말 시계 오차\n",
        "        elif k == \"qty_diff\":\n",
        "            ev.update(d=float(rng.uniform(0.0, 0.8)))\n",
        "        else:\n",
        "            ev.update(p=0.5 if rng.random() < .6 else 0.8, u=rng.random(8))\n",
        "        out.append(ev)\n",
        "    return out\n",
        "\n",
        "\n",
        "def decide(ev, P):\n",
        "    k = ev[\"kind\"]\n",
        "    if k in (\"dup_resend\", \"distinct_similar\", \"dup_reinput\"):\n",
        "        if P[\"dup\"] == \"id\":\n",
        "            return \"skip\" if ev[\"same_id\"] else \"apply\"\n",
        "        if P[\"dup\"] == \"id+content\":\n",
        "            return \"skip\" if ev[\"same_id\"] or ev[\"dt\"] <= P[\"win\"] else \"apply\"\n",
        "        return \"skip\" if ev[\"dt\"] <= P[\"win\"] else \"apply\"                  # 내용 비교: 같은 품목·수량이 시간창 안이면 중복\n",
        "    if k in (\"meta_conflict\", \"meta_causal\"):\n",
        "        if P[\"conflict\"] == \"vv\" and ev[\"vv\"]:\n",
        "            return \"hitl\" if k == \"meta_conflict\" else \"pick_b\"\n",
        "        if abs(ev[\"ta\"] - ev[\"tb\"]) <= P[\"tw\"]:\n",
        "            return \"hitl\"\n",
        "        return \"pick_b\" if ev[\"tb\"] > ev[\"ta\"] else \"pick_a\"                # 마지막 기록 우선(LWW)\n",
        "    if k == \"qty_diff\":\n",
        "        return \"auto\" if ev[\"d\"] < P[\"thr\"] else \"hitl\"\n",
        "    return \"applied\" if (ev[\"u\"][:1 + P[\"retries\"]] < ev[\"p\"]).any() else \"failed\"\n",
        "\n",
        "\n",
        "def truth(ev):\n",
        "    k = ev[\"kind\"]\n",
        "    if k == \"qty_diff\":\n",
        "        return \"auto\" if ev[\"d\"] < 0.25 else \"hitl\"\n",
        "    return {\"dup_resend\": \"skip\", \"dup_reinput\": \"skip\", \"distinct_similar\": \"apply\", \"meta_conflict\": \"hitl\", \"meta_causal\": \"pick_b\", \"send_fail\": \"applied\"}[k]\n",
        "\n",
        "\n",
        "def evaluate(evs, P):\n",
        "    ok = [decide(e, P) == truth(e) for e in evs]\n",
        "    return np.mean(ok), pd.Series([CAUSE[e[\"kind\"]] for e, o in zip(evs, ok) if not o]).value_counts(), ok\n",
        "\n",
        "\n",
        "P0 = {\"dup\": \"content\", \"win\": 2.0, \"conflict\": \"time\", \"tw\": 4.0, \"thr\": 0.3, \"retries\": 2}           # 초기 운용 규칙(M3 단계 설정)\n",
        "aar, test = make_events(31, 1200), make_events(77, 3000)\n",
        "a0, causes, ok0 = evaluate(aar, P0)\n",
        "print(f\"[AAR 사례] {len(aar)}건, 현행 규칙 정확도 {a0:.1%}, 오류 {int(len(aar) * (1 - a0))}건\")\n",
        "print(\"[원인 분류] 원인 / 건수 / 비율\")\n",
        "for c, n in causes.items():\n",
        "    print(f\"  {c:28s} {n:4d}건 {n / (len(aar) * (1 - a0)):5.1%}\")\n",
        "\n",
        "# ---- 재튜닝: 사례 종류가 서로 다르므로 파라미터마다 AAR 사례에서 따로 탐색(재시도는 횟수당 지연 비용 0.2%p 가산) ----\n",
        "def best(param, grid, kinds, cost=lambda v: 0.0):\n",
        "    sub = [e for e in aar if e[\"kind\"] in kinds]; sc = []\n",
        "    for v in grid:\n",
        "        P = {**P0, **(v if isinstance(v, dict) else {param: v})}; sc.append((np.mean([decide(e, P) == truth(e) for e in sub]) - cost(v), v))\n",
        "    return max(sc, key=lambda x: x[0]), sc\n",
        "(_, dup_best), _ = best(\"dup\", [{\"dup\": \"content\", \"win\": w} for w in (0.0, 0.25, 0.5, 1, 2, 4)] + [{\"dup\": \"id+content\", \"win\": w} for w in (0.25, 0.5, 1, 2, 4)] + [{\"dup\": \"id\"}], (\"dup_resend\", \"distinct_similar\", \"dup_reinput\"))\n",
        "(_, cf_best), _ = best(\"conflict\", [{\"conflict\": \"time\", \"tw\": w} for w in (0.25, 0.5, 1, 2, 4)] + [{\"conflict\": \"vv\", \"tw\": w} for w in (0.5, 1, 2, 4, 8)], (\"meta_conflict\", \"meta_causal\"))\n",
        "(_, thr_best), _ = best(\"thr\", [0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5], (\"qty_diff\",))\n",
        "(_, rt_best), _ = best(\"retries\", [1, 2, 3, 4, 5, 6], (\"send_fail\",), cost=lambda v: 0.002 * v)\n",
        "P1 = {**P0, **dup_best, **cf_best, \"thr\": thr_best, \"retries\": rt_best}\n",
        "\n",
        "a_aar, rec, ok1 = evaluate(aar, P1)\n",
        "a_old_t, c_old_t, _ = evaluate(test, P0)\n",
        "a_new_t, c_new_t, _ = evaluate(test, P1)\n",
        "rec_n = int(sum((not o0) and (not o1) for o0, o1 in zip(ok0, ok1)))\n",
        "print(f\"\\n[재튜닝 결과] 중복 판정 {P0['dup']}(창 {P0['win']}h) -> {P1['dup']}{'' if P1['dup'] == 'id' else '(창 ' + str(P1['win']) + 'h)'}, 충돌 탐지 {P0['conflict']}(창 {P0['tw']}h) -> {P1['conflict']}(구형 기록은 시간창 {P1['tw']}h), 자동 병합 임계 {P0['thr']} -> {P1['thr']}, 재시도 {P0['retries']} -> {P1['retries']}회\")\n",
        "print(f\"AAR 사례 재실행: 정확도 {a0:.1%} -> {a_aar:.1%}, 같은 사례에서 오류 재발 {rec_n}건 (목표 0건: {'달성' if rec_n == 0 else '미달'})\")\n",
        "print(f\"별도 시험 3,000건: {a_old_t:.1%} -> {a_new_t:.1%} ({(a_new_t - a_old_t) * 100:+.1f}%p, 목표 +10%p 이상: {'달성' if a_new_t - a_old_t >= .10 else '미달'})\")\n",
        "print(\"시험 3,000건 남은 오류:\", {c: int(n) for c, n in c_new_t.items()} or \"없음\")\n",
        "tt = pd.DataFrame({\"before\": c_old_t, \"after\": c_new_t}).fillna(0).astype(int)\n",
        "\n",
        "# ---- 회귀 테스트: 규칙 단위 시험 8개를 현행/재튜닝 규칙에 각각 실행 ----\n",
        "T = [(\"같은 ID 재전송(시각 3h 차이)은 건너뜀\", {\"kind\": \"dup_resend\", \"same_id\": True, \"dt\": 3.0, \"vv\": True}, \"skip\"), (\"같은 ID 재전송(시각 동일)은 건너뜀\", {\"kind\": \"dup_resend\", \"same_id\": True, \"dt\": 0.0, \"vv\": True}, \"skip\"),\n",
        "     (\"다른 ID의 같은 수량 거래는 반영\", {\"kind\": \"distinct_similar\", \"same_id\": False, \"dt\": 0.2, \"vv\": True}, \"apply\"), (\"버전 벡터 동시 수정은 군수관 검토\", {\"kind\": \"meta_conflict\", \"ta\": 100, \"tb\": 130, \"vv\": True}, \"hitl\"),\n",
        "     (\"버전 벡터 선후 관계는 후속 값 채택(시계 역전)\", {\"kind\": \"meta_causal\", \"ta\": 101.0, \"tb\": 100.5, \"vv\": True}, \"pick_b\"), (\"차이 10%는 자동 병합\", {\"kind\": \"qty_diff\", \"d\": 0.10, \"vv\": True}, \"auto\"),\n",
        "     (\"차이 30%는 군수관 검토\", {\"kind\": \"qty_diff\", \"d\": 0.30, \"vv\": True}, \"hitl\"), (\"통신불량 전송 3회 안에 성공\", {\"kind\": \"send_fail\", \"p\": 0.5, \"u\": np.array([.9, .8, .1, 0, 0, 0, 0, 0]), \"vv\": True}, \"applied\")]\n",
        "print(\"\\n[회귀 테스트 8개] 현행 / 재튜닝\")\n",
        "p0 = [decide(e, P0) == t for _, e, t in T]; p1 = [decide(e, P1) == t for _, e, t in T]\n",
        "for (nm, _, _), x, y in zip(T, p0, p1):\n",
        "    print(f\"  {nm:34s} {'통과' if x else '실패'} / {'통과' if y else '실패'}\")\n",
        "regress = sum(a and not b for a, b in zip(p0, p1))\n",
        "print(f\"통과 {sum(p0)}/8 -> {sum(p1)}/8, 기존 통과 항목이 깨진 회귀 {regress}건 ({'전체 통과' if all(p1) else '실패 있음'})\")\n",
        "\n",
        "# ---- 산출물: 튜닝 파라미터 YAML 문자열(configs/dat10_tuned_params.yaml 내용) ----\n",
        "yml = \"\\n\".join([\"sync_rules:\", f\"  dedup_key: {P1['dup']}\", f\"  dedup_window_h: {P1['win'] if P1['dup'] != 'id' else 0}\", f\"  conflict_detect: {P1['conflict']}\", f\"  legacy_time_window_h: {P1['tw']}\", f\"  auto_merge_diff_max: {P1['thr']}\", f\"  send_retries: {P1['retries']}\", \"  hitl_required_when: [conflict, diff_ge_0.25]\"])\n",
        "print(\"\\n[configs/dat10_tuned_params.yaml]\"); print(yml)\n",
        "print(\"[지속적 학습 연계] 이번 AAR 오류 유형을 사례 은행에 누적 -> 월 1회 새 사례로 파라미터 재탐색 -> 회귀 테스트 8개 통과 시에만 후보 반영 -> 군수관 승인 후 배포(자동 배포 금지)\")\n",
        "print(\"※ 사례는 가상 생성이며 시험 분포가 튜닝 분포와 같아 향상폭이 실제보다 낙관적이다. 규칙 변경은 군수관 승인과 회귀 테스트를 거친다.\")\n",
        "\n",
        "fig, ax = plt.subplots(1, 2, figsize=(10, 3.8))\n",
        "ax[0].bar([\"before\", \"after\"], [a_old_t * 100, a_new_t * 100], color=[\"tab:gray\", \"tab:blue\"]); ax[0].set_ylim(50, 100); ax[0].set_ylabel(\"accuracy (%)\"); ax[0].set_title(\"Test accuracy (3,000 cases)\")\n",
        "x = np.arange(len(tt)); ax[1].bar(x - .2, tt[\"before\"], .4, label=\"before\", color=\"tab:gray\"); ax[1].bar(x + .2, tt[\"after\"], .4, label=\"after\", color=\"tab:blue\")\n",
        "ax[1].set_xticks(x); ax[1].set_xticklabels([c.split()[0] for c in tt.index]); ax[1].set_title(\"Errors by cause\"); ax[1].legend(); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[AAR 사례] 1200건, 현행 규칙 정확도 45.5%, 오류 653건\n",
        "[원인 분류] 원인 / 건수 / 비율\n",
        "  D4 동시 수정 놓침(조용한 덮어쓰기)         193건 29.5%\n",
        "  D3 후속 수정의 승자 오판·불필요 경보        177건 27.1%\n",
        "  D1 다른 거래를 중복으로 판정             139건 21.3%\n",
        "  D2 재전송 중복 놓침                   95건 14.5%\n",
        "  D2b 재입력(새 ID) 중복 놓침            25건  3.8%\n",
        "  D6 재시도 부족 누락                   17건  2.6%\n",
        "  D5 자동 병합 임계 부적절                 8건  1.2%\n",
        "\n",
        "[재튜닝 결과] 중복 판정 content(창 2.0h) -> id, 충돌 탐지 time(창 4.0h) -> vv(구형 기록은 시간창 0.5h), 자동 병합 임계 0.3 -> 0.25, 재시도 2 -> 6회\n",
        "AAR 사례 재실행: 정확도 45.5% -> 93.7%, 같은 사례에서 오류 재발 49건 (목표 0건: 미달)\n",
        "별도 시험 3,000건: 44.7% -> 93.0% (+48.3%p, 목표 +10%p 이상: 달성)\n",
        "시험 3,000건 남은 오류: {'D2b 재입력(새 ID) 중복 놓침': 164, 'D4 동시 수정 놓침(조용한 덮어쓰기)': 32, 'D3 후속 수정의 승자 오판·불필요 경보': 11, 'D6 재시도 부족 누락': 2}\n",
        "\n",
        "[회귀 테스트 8개] 현행 / 재튜닝\n",
        "  같은 ID 재전송(시각 3h 차이)은 건너뜀           실패 / 통과\n",
        "  같은 ID 재전송(시각 동일)은 건너뜀              통과 / 통과\n",
        "  다른 ID의 같은 수량 거래는 반영                실패 / 통과\n",
        "  버전 벡터 동시 수정은 군수관 검토                실패 / 통과\n",
        "  버전 벡터 선후 관계는 후속 값 채택(시계 역전)        실패 / 통과\n",
        "  차이 10%는 자동 병합                      통과 / 통과\n",
        "  차이 30%는 군수관 검토                     통과 / 통과\n",
        "  통신불량 전송 3회 안에 성공                   통과 / 통과\n",
        "통과 4/8 -> 8/8, 기존 통과 항목이 깨진 회귀 0건 (전체 통과)\n",
        "\n",
        "[configs/dat10_tuned_params.yaml]\n",
        "sync_rules:\n",
        "  dedup_key: id\n",
        "  dedup_window_h: 0\n",
        "  conflict_detect: vv\n",
        "  legacy_time_window_h: 0.5\n",
        "  auto_merge_diff_max: 0.25\n",
        "  send_retries: 6\n",
        "  hitl_required_when: [conflict, diff_ge_0.25]\n",
        "[지속적 학습 연계] 이번 AAR 오류 유형을 사례 은행에 누적 -> 월 1회 새 사례로 파라미터 재탐색 -> 회귀 테스트 8개 통과 시에만 후보 반영 -> 군수관 승인 후 배포(자동 배포 금지)\n",
        "※ 사례는 가상 생성이며 시험 분포가 튜닝 분포와 같아 향상폭이 실제보다 낙관적이다. 규칙 변경은 군수관 승인과 회귀 테스트를 거친다.\n",
        "```\n",
        "</details>\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "\n",
        "## 🎯 미션 10. 오프라인 동기화 야전 재고 관리 SaaS 최종 패키징 및 야전 배포 가이드 완성\n",
        "**소요시간**: 약 3.5시간\n",
        "\n",
        "### 📋 미션 배경 및 수행 지침\n",
        "#### 미션 배경 및 당위성\n",
        "\n",
        "오프라인 동기화 야전 재고 관리 SaaS 패키징 및 배포 가이드를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **입력 검증·서명 점검·중복 제거·PN-카운터 병합·충돌 탐지·공급망 재난 징후 마이닝·검토 큐·감사 로그·브리핑을 묶은 오프라인 동기화 야전 재고 관리 SaaS 패키지와 배포 가이드를 완성한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
        "\n",
        "이 문제를 해결하기 위해 본 미션에서는 **앞 미션의 기능을 통합 파이프라인으로 묶어 설치 스크립트 5단계 점검, 시드 4종 End-to-End 실행(업체 1곳 공급 중단, 노드 1곳 3일치 미동기, 변조 20·깨진 줄 5·중복 60건 주입), 자체 점검 11항목과 함수 커버리지 수행하고 분석하는** 실습을 체계적으로 수행합니다.\n",
        "\n",
        "학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
        "\n",
        "**이번 결과물:** `src/main_dat10.py, README.md, docs/dat10_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": [
        "\"\"\"오프라인 입력 검증·서명 점검·중복 제거, PN-카운터 병합, 버전 벡터 충돌 탐지, 공급망 재난 징후(업체 공급 감소·재고 소진 임박) 마이닝, 사람 검토 큐(PENDING)와 해시 체인 감사 로그, 설치 스크립트·매뉴얼·브리핑까지 오프라인 동기화 야전 재고 관리 SaaS 전체를 한 파일로 통합 실행하고 E2E 성공률·자체 점검·함수 커버리지를 계산한다.\"\"\"\n",
        "import hashlib\n",
        "import hmac\n",
        "import json\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",
        "# 가상 보급소 5곳이 품목 12종의 입고·불출·조정 트랜잭션을 오프라인으로 기록하고 나중에 동기화한다.\n",
        "# 공급망 재난 징후(업체 납기 지연, 품목 재고 급감)는 동기화된 자료를 마이닝해 식별한다.\n",
        "# ---------------------------------------------------------------------------\n",
        "NODES = [\"보급소A\", \"보급소B\", \"보급소C\", \"보급소D\", \"보급소E\"]\n",
        "SUPPLIERS = [\"가상업체1\", \"가상업체2\", \"가상업체3\", \"가상업체4\", \"가상업체5\", \"가상업체6\"]\n",
        "CATS = [\"일반보급\", \"정비부품\", \"의무소모\"]\n",
        "ITEMS = [{\"item\": f\"I{i + 1:02d}\", \"cat\": CATS[i % 3], \"supplier\": SUPPLIERS[i % 6], \"lead_days\": 3 + (i * 5) % 12, \"use\": 4 + (i * 7) % 11, \"min_level\": 30 + 10 * (i % 5)} for i in range(12)]\n",
        "ITEM_IDS = [x[\"item\"] for x in ITEMS]\n",
        "CTXS = [\"평시\", \"야간\", \"통신불량\"]\n",
        "\n",
        "\n",
        "def make_txns(seed=7, days=7, per_node_day=60, nodes=None):\n",
        "    \"\"\"노드별 오프라인 트랜잭션 목록. id는 노드-일련번호(멱등 키), t는 시각(시간 단위), ctx는 입력 상황.\"\"\"\n",
        "    rng = np.random.default_rng(seed)\n",
        "    out = []\n",
        "    for n in (nodes or NODES):\n",
        "        seq = 0\n",
        "        for k in range(days * per_node_day):\n",
        "            seq += 1\n",
        "            it = ITEMS[int(rng.integers(len(ITEMS)))]\n",
        "            t = float(rng.uniform(0, 24 * days))\n",
        "            ctx = \"야간\" if (t % 24 >= 20 or t % 24 < 5) else \"평시\"\n",
        "            if rng.random() < 0.12:\n",
        "                ctx = \"통신불량\"\n",
        "            r = rng.random()\n",
        "            if r < 0.22:\n",
        "                op, q = \"in\", int(it[\"use\"] * rng.uniform(4, 9))\n",
        "            elif r < 0.95:\n",
        "                op, q = \"out\", int(max(1, rng.normal(it[\"use\"] * 1.5, 3)))\n",
        "            else:\n",
        "                op, q = \"adjust\", int(rng.integers(-5, 6))\n",
        "            out.append({\"id\": f\"{n}-{seq:05d}\", \"node\": n, \"item\": it[\"item\"], \"op\": op, \"qty\": q, \"t\": round(t, 2), \"ctx\": ctx, \"supplier\": it[\"supplier\"]})\n",
        "    return out\n",
        "\n",
        "\n",
        "def stock_of(txns, base=500):\n",
        "    \"\"\"품목별 재고 = 기초 재고 + 입고 - 불출 + 조정 (id 중복은 한 번만 반영).\"\"\"\n",
        "    st, seen = {i: base for i in ITEM_IDS}, set()\n",
        "    for x in txns:\n",
        "        if x[\"id\"] in seen:\n",
        "            continue\n",
        "        seen.add(x[\"id\"])\n",
        "        st[x[\"item\"]] += x[\"qty\"] if x[\"op\"] in (\"in\", \"adjust\") else -x[\"qty\"]\n",
        "    return st\n",
        "\n",
        "KEYS = {n: hashlib.sha256(f\"demo-key-{n}\".encode()).digest() for n in NODES}            # [교육용가상자료] 데모 키\n",
        "CONF = {\"auto_merge_diff_max\": 0.25, \"supply_ratio_cut\": 0.3, \"supplier_flag_frac\": 0.5, \"queue_min_risk\": 40}\n",
        "MANUAL = {\n",
        "    \"1. 목적과 범위\": \"통신이 끊긴 보급소 5곳의 재고 입출고를 오프라인으로 기록하고 복구 후 서버와 병합하며, 병합된 자료에서 공급 감소·재고 소진 임박 징후를 찾아 군수관에게 보여 주는 판단 보조 도구다.\",\n",
        "    \"2. 설치와 실행\": \"install.sh 한 번으로 환경 점검, 의존 패키지 확인, 자체 시험, 데모 실행을 순서대로 수행한다.\",\n",
        "    \"3. 입력 데이터\": \"보급소별 트랜잭션 줄(ID|품목|구분|수량|시각|상황). 범위·형식·서명 오류와 중복은 격리하거나 거부하고 사유를 남긴다.\",\n",
        "    \"4. 동기화와 충돌\": \"재고 수량은 PN-카운터로 자동 병합한다. 품목 정보 동시 수정은 버전 벡터로 탐지해 군수관 검토 큐에 올린다.\",\n",
        "    \"5. 공급망 재난 징후\": \"업체별 최근 3일 입고량이 첫 3일 대비 급감하거나 소진 예상 일수가 납기보다 짧은 품목을 묶어 징후 후보로 표시한다.\",\n",
        "    \"6. 사람 검토와 감사 로그\": \"모든 후보는 PENDING으로 시작하고 군수관·지휘관이 처리한다. 모든 행위는 해시 체인에 남는다.\",\n",
        "    \"7. 한계와 확장\": \"가상 자료 기반이며 실제 통신·저장 장치·키 관리는 별도 체계가 필요하다. 인접 체계 연동 API는 차기 과제다.\",\n",
        "}\n",
        "INSTALL_SH = \"\"\"#!/bin/sh\n",
        "# [교육용가상자료] dat10 설치 스크립트 예시\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 matplotlib)\n",
        "python3 -m pytest tests -q                                           # 4) 자체 시험\n",
        "python3 src/main_dat10.py --demo                                     # 5) 데모 실행(가상 자료)\n",
        "\"\"\"\n",
        "\n",
        "\n",
        "def sign(x):\n",
        "    return hmac.new(KEYS[x[\"node\"]], f\"{x['id']}|{x['item']}|{x['op']}|{x['qty']}|{x['t']}\".encode(), \"sha256\").hexdigest()[:16]\n",
        "\n",
        "\n",
        "def to_line(x):\n",
        "    return f\"{x['id']}|{x['item']}|{x['op']}|{x['qty']}|{x['t']}|{x['ctx']}|{x['sig']}\"\n",
        "\n",
        "\n",
        "def parse_line(s):\n",
        "    \"\"\"입력 줄 -> 레코드. 형식이 틀리면 ValueError.\"\"\"\n",
        "    p = s.split(\"|\")\n",
        "    if len(p) != 7:\n",
        "        raise ValueError(\"필드 수 오류\")\n",
        "    return {\"id\": p[0], \"node\": p[0].split(\"-\")[0], \"item\": p[1], \"op\": p[2], \"qty\": int(p[3]), \"t\": float(p[4]), \"ctx\": p[5], \"sig\": p[6]}\n",
        "\n",
        "\n",
        "def validate(x):\n",
        "    \"\"\"범위·논리 오류 사유 문자열 또는 None.\"\"\"\n",
        "    if x[\"node\"] not in NODES:\n",
        "        return \"미등록 보급소\"\n",
        "    if x[\"item\"] not in ITEM_IDS or x[\"op\"] not in (\"in\", \"out\", \"adjust\") or x[\"ctx\"] not in CTXS:\n",
        "        return \"코드 오류\"\n",
        "    if not (0 <= x[\"t\"] <= 24 * 7 + 1) or not (-50 <= x[\"qty\"] <= 2000) or (x[\"op\"] != \"adjust\" and x[\"qty\"] <= 0):\n",
        "        return \"수량·시각 범위 오류\"\n",
        "    return None\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 Store:\n",
        "    \"\"\"서버 측 저장소(메모리): PN-카운터, 중복 제거 집합, 격리 목록.\"\"\"\n",
        "    def __init__(self):\n",
        "        self.P = {i: {} for i in ITEM_IDS}; self.N = {i: {} for i in ITEM_IDS}; self.ids = set(); self.tx = []; self.quar = []; self.dups = 0\n",
        "\n",
        "    def stock(self, base=500):\n",
        "        return {i: base + sum(self.P[i].values()) - sum(self.N[i].values()) for i in ITEM_IDS}\n",
        "\n",
        "\n",
        "def ingest(st, lines):\n",
        "    \"\"\"줄 묶음을 파싱·검증·서명 점검·중복 제거 후 반영한다. 결과 건수를 돌려준다.\"\"\"\n",
        "    r = {\"ok\": 0, \"dup\": 0, \"bad\": 0}\n",
        "    for s in lines:\n",
        "        try:\n",
        "            x = parse_line(s)\n",
        "        except ValueError:\n",
        "            st.quar.append((\"파싱 오류\", s[:20])); r[\"bad\"] += 1; continue\n",
        "        why = validate(x)\n",
        "        if why is None and not hmac.compare_digest(sign(x), x[\"sig\"]):\n",
        "            why = \"서명 불일치\"\n",
        "        if why:\n",
        "            st.quar.append((why, x[\"id\"])); r[\"bad\"] += 1; continue\n",
        "        if x[\"id\"] in st.ids:\n",
        "            st.dups += 1; r[\"dup\"] += 1; continue\n",
        "        st.ids.add(x[\"id\"]); st.tx.append(x)\n",
        "        d = st.P if x[\"op\"] == \"in\" or (x[\"op\"] == \"adjust\" and x[\"qty\"] > 0) else st.N\n",
        "        d[x[\"item\"]][x[\"node\"]] = d[x[\"item\"]].get(x[\"node\"], 0) + abs(x[\"qty\"]); r[\"ok\"] += 1\n",
        "    return r\n",
        "\n",
        "\n",
        "def dominated(a, b):\n",
        "    return all(a.get(k, 0) <= b.get(k, 0) for k in set(a) | set(b)) and a != b\n",
        "\n",
        "\n",
        "def detect_conflict(A, B):\n",
        "    \"\"\"버전 벡터 + 값 비교: 같은 필드를 서로 모르는 채 다른 값으로 고쳤을 때만 충돌.\"\"\"\n",
        "    return A[\"field\"] == B[\"field\"] and A[\"value\"] != B[\"value\"] and not dominated(A[\"vv\"], B[\"vv\"]) and not dominated(B[\"vv\"], A[\"vv\"])\n",
        "\n",
        "\n",
        "def make_conflicts(rng, n=10):\n",
        "    out = []\n",
        "    for k in range(n):\n",
        "        it = ITEMS[int(rng.integers(12))]; va = it[\"min_level\"]; vb = int(va * rng.choice([1.1, 1.5, 0.6]))\n",
        "        conc = k % 2 == 0\n",
        "        A = {\"item\": it[\"item\"], \"field\": \"min_level\", \"value\": va, \"vv\": {\"A\": 3, \"B\": 1}}\n",
        "        B = {\"item\": it[\"item\"], \"field\": \"min_level\", \"value\": vb, \"vv\": {\"A\": 2, \"B\": 4} if conc else {\"A\": 3, \"B\": 2}}\n",
        "        out.append((conc, A, B))\n",
        "    return out\n",
        "\n",
        "\n",
        "def disruption_scan(tx, t_end=168.0, win=72.0):\n",
        "    \"\"\"업체별 공급 감소(최근 3일 입고량 / 첫 3일 입고량)와 소진 예상 일수를 보고 공급망 재난 징후 후보 업체를 찾는다.\"\"\"\n",
        "    df = pd.DataFrame(tx)\n",
        "    ins = df[df[\"op\"] == \"in\"]; outs = df[df[\"op\"] == \"out\"]\n",
        "    rows = []\n",
        "    for it in ITEMS:\n",
        "        a = ins[(ins[\"item\"] == it[\"item\"]) & (ins[\"t\"] < win)][\"qty\"].sum(); b = ins[(ins[\"item\"] == it[\"item\"]) & (ins[\"t\"] >= t_end - win)][\"qty\"].sum()\n",
        "        o = outs[(outs[\"item\"] == it[\"item\"]) & (outs[\"t\"] >= t_end - win)][\"qty\"].sum() / (win / 24)\n",
        "        rows.append({\"supplier\": it[\"supplier\"], \"item\": it[\"item\"], \"ratio\": b / max(a, 1), \"out_day\": o, \"lead\": it[\"lead_days\"]})\n",
        "    r = pd.DataFrame(rows); r[\"drop\"] = r[\"ratio\"] < CONF[\"supply_ratio_cut\"]\n",
        "    g = r.groupby(\"supplier\").agg(frac=(\"drop\", \"mean\"), ratio=(\"ratio\", \"min\"))\n",
        "    return g[g[\"frac\"] >= CONF[\"supplier_flag_frac\"]].reset_index().to_dict(\"records\"), r\n",
        "\n",
        "\n",
        "def dashboard_view(st, pending, t):\n",
        "    off = sum(1 for n in NODES if pending.get(n, 0) > 0)\n",
        "    stk = st.stock()\n",
        "    return {\"nodes_unsynced\": off, \"pending_tx\": sum(pending.values()), \"low_stock\": sum(v < 100 for v in stk.values()), \"alerts\": len(st.quar)}\n",
        "\n",
        "\n",
        "def run_pipeline(seed, disrupt=None, offline_node=None, noise=True):\n",
        "    \"\"\"E2E: 생성 -> (공급 중단·오프라인 주입) -> 줄 변환 -> 반영 -> 충돌 탐지 -> 징후 마이닝 -> 검토 큐 + 감사 로그.\"\"\"\n",
        "    rng = np.random.default_rng(seed); tx = make_txns(seed, 7, per_node_day=60)\n",
        "    if disrupt:\n",
        "        tx = [x for x in tx if not (x[\"supplier\"] == disrupt and x[\"op\"] == \"in\" and x[\"t\"] >= 72)]\n",
        "    for x in tx:\n",
        "        x[\"sig\"] = sign(x)\n",
        "    lines, pending = [], {}\n",
        "    for x in tx:\n",
        "        if offline_node and x[\"node\"] == offline_node and x[\"t\"] >= 96:\n",
        "            pending[x[\"node\"]] = pending.get(x[\"node\"], 0) + 1; continue                 # 통신 단절 노드의 마지막 3일분은 아직 미동기\n",
        "        lines.append(to_line(x))\n",
        "    n_bad = 0\n",
        "    if noise:\n",
        "        idx = rng.permutation(len(lines)); n_bad = 25\n",
        "        for i in idx[:20]:\n",
        "            p = lines[i].split(\"|\"); p[3] = str(int(p[3]) + 9); lines[i] = \"|\".join(p)                  # 서명 후 수량 변조\n",
        "        for i in idx[20:25]:\n",
        "            lines[i] = lines[i].rsplit(\"|\", 2)[0]                                                       # 줄 깨짐\n",
        "        lines += [lines[i] for i in idx[25:85]]                                                         # 정상 줄 60건 중복 재전송\n",
        "    st = Store(); audit = Chain(); res = {\"ok\": 0, \"dup\": 0, \"bad\": 0}\n",
        "    for k in range(0, len(lines), 100):\n",
        "        r = ingest(st, lines[k:k + 100]); res = {a: res[a] + r[a] for a in res}\n",
        "    audit.add(event=\"SYNC\", ok=res[\"ok\"], dup=res[\"dup\"], bad=res[\"bad\"])\n",
        "    conf = make_conflicts(rng); hit = [detect_conflict(A, B) for _, A, B in conf]; ctp = sum(h and c for h, (c, _, _) in zip(hit, conf)); cfp = sum(h and not c for h, (c, _, _) in zip(hit, conf))\n",
        "    flagged, items = disruption_scan(st.tx)\n",
        "    queue = []\n",
        "    for (c, A, B), h in zip(conf, hit):\n",
        "        if h:\n",
        "            queue.append({\"kind\": \"충돌\", \"ref\": A[\"item\"], \"risk\": 60, \"status\": \"PENDING\"}); audit.add(event=\"QUEUE\", kind=\"충돌\", item=A[\"item\"], status=\"PENDING\")\n",
        "    for f in flagged:\n",
        "        queue.append({\"kind\": \"공급 징후\", \"ref\": f[\"supplier\"], \"risk\": int(100 * (1 - min(f[\"ratio\"], 1))), \"status\": \"PENDING\"}); audit.add(event=\"QUEUE\", kind=\"공급 징후\", supplier=f[\"supplier\"], status=\"PENDING\")\n",
        "    return {\"st\": st, \"res\": res, \"n_bad\": n_bad, \"queue\": queue, \"audit\": audit, \"flagged\": [f[\"supplier\"] for f in flagged], \"ctp\": ctp, \"cfp\": cfp, \"nconf\": sum(c for c, _, _ in conf),\n",
        "            \"pending\": pending, \"valid\": [x for x in tx if x[\"id\"] in st.ids], \"view\": dashboard_view(st, pending, 168)}\n",
        "\n",
        "\n",
        "def self_tests():\n",
        "    out = {}\n",
        "    r = run_pipeline(7)\n",
        "    out[\"정상 자료 통합 실행(오류 주입 포함)\"] = r[\"res\"][\"ok\"] > 2000 and len(r[\"st\"].quar) == r[\"n_bad\"]\n",
        "    truth = stock_of([x for x in make_txns(7, 7, 60) if x[\"id\"] in r[\"st\"].ids])\n",
        "    out[\"서버 재고 = 반영된 레코드 합산\"] = r[\"st\"].stock() == truth\n",
        "    st = Store(); lines = [to_line({**x, \"sig\": sign(x)}) for x in make_txns(3, 2, 20)]; a = ingest(st, lines); b = ingest(st, lines)\n",
        "    out[\"같은 묶음 재전송은 상태를 바꾸지 않음(멱등)\"] = a[\"ok\"] == len(lines) and b[\"ok\"] == 0 and b[\"dup\"] == len(lines)\n",
        "    out[\"파싱·범위 오류 줄 격리\"] = ingest(Store(), [\"깨진줄\", \"보급소A-00001|I99|in|5|1.0|평시|0\"])[\"bad\"] == 2\n",
        "    out[\"충돌 탐지(동시 수정만 충돌)\"] = r[\"ctp\"] == r[\"nconf\"] and r[\"cfp\"] == 0\n",
        "    r2 = run_pipeline(17, disrupt=\"가상업체3\", offline_node=\"보급소B\")\n",
        "    out[\"공급 중단 업체 탐지 + 오프라인 노드로 오경보 없음\"] = r2[\"flagged\"] == [\"가상업체3\"]\n",
        "    out[\"정상 자료에서 공급 징후 오경보 0\"] = r[\"flagged\"] == []\n",
        "    out[\"모든 검토 건이 PENDING(자동 확정 없음)\"] = all(q[\"status\"] == \"PENDING\" for q in r2[\"queue\"]) and len(r2[\"queue\"]) > 0\n",
        "    r2[\"audit\"].log[1][\"ok\"] = 1\n",
        "    out[\"감사 로그 변조 탐지\"] = r2[\"audit\"].verify() == 1\n",
        "    out[\"대시보드 요약 4개 지표\"] = set(r2[\"view\"]) == {\"nodes_unsynced\", \"pending_tx\", \"low_stock\", \"alerts\"} and r2[\"view\"][\"pending_tx\"] > 0\n",
        "    out[\"매뉴얼 7개 장·설치 스크립트 5단계\"] = all(len(v) > 20 for v in MANUAL.values()) and all(k in INSTALL_SH for k in (\"requirements\", \"pytest\", \"--demo\", \"venv\", \"version_info\"))\n",
        "    return out\n",
        "\n",
        "\n",
        "def briefing(r):\n",
        "    q = r[\"queue\"]; v = r[\"view\"]\n",
        "    return [f\"1. 동기화: 반영 {r['res']['ok']}건, 중복 거부 {r['res']['dup']}건, 격리 {len(r['st'].quar)}건. 미동기 {v['pending_tx']}건({v['nodes_unsynced']}개 보급소).\",\n",
        "            f\"2. 검토 큐 {len(q)}건(충돌 {sum(x['kind'] == '충돌' for x in q)}, 공급 징후 {sum(x['kind'] == '공급 징후' for x in q)}): 모두 PENDING.\",\n",
        "            f\"3. 공급망 재난 징후 후보 업체: {', '.join(r['flagged']) or '없음'} (최근 3일 입고량이 첫 3일의 30% 미만).\",\n",
        "            \"4. 병합 확정과 보급 조치는 군수관·지휘관이 한다. 본 도구는 가상 자료 기반 판단 보조이다.\"]\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",
        "tests = self_tests()\n",
        "e2e = []\n",
        "for s, dis, off in ((7, \"가상업체1\", \"보급소A\"), (17, \"가상업체3\", \"보급소B\"), (27, \"가상업체5\", None), (37, \"가상업체6\", \"보급소E\")):\n",
        "    try:\n",
        "        r = run_pipeline(s, disrupt=dis, offline_node=off); truth = stock_of(r[\"valid\"])\n",
        "        e2e.append((s, True, dis, r[\"flagged\"], r[\"res\"], len(r[\"st\"].quar), r[\"n_bad\"], r[\"st\"].stock() == truth, r[\"audit\"].verify() is None, len(r[\"queue\"])))\n",
        "    except Exception as e:\n",
        "        e2e.append((s, False, dis, [], {}, 0, 0, False, False, 0))\n",
        "brief = briefing(run_pipeline(7, disrupt=\"가상업체1\", offline_node=\"보급소A\"))\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개, 업체 1곳 공급 중단, 노드 1곳 3일치 미동기, 변조 20·깨진 줄 5·중복 60건 주입]\")\n",
        "for s, ok, dis, fl, rs, nq, nb, st_ok, chain, nqueue in e2e:\n",
        "    print(f\"  seed {s}: {'성공' if ok else '실패'}, 반영 {rs.get('ok')}건·중복 {rs.get('dup')}건·격리 {nq}/{nb}건, 재고 일치 {'예' if st_ok else '아니오'}, 징후 {fl} (주입 {dis}), 검토 큐 {nqueue}건, 감사 로그 {'정상' if chain else '이상'}\")\n",
        "hit = sum(x[2] in x[3] for x in e2e); fa = sum(len([f for f in x[3] if f != x[2]]) for x in e2e)\n",
        "print(f\"E2E 성공률 {np.mean([x[1] for x in e2e]):.0%} (목표 100%), 공급 중단 업체 탐지 {hit}/{len(e2e)}, 다른 업체 오경보 {fa}건\")\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",
        "print(\"\\n[군수관 브리핑 요약]\")\n",
        "for b in brief:\n",
        "    print(b)\n",
        "print(\"※ 가상 자료·메모리 모의이며 실제 통신·저장 장치·키 관리·인접 체계 연동은 포함하지 않는다. 함수 커버리지는 최상위 함수 호출 기준이다.\")\n",
        "\n",
        "r = run_pipeline(17, disrupt=\"가상업체3\", offline_node=\"보급소B\"); d, tab = disruption_scan(r[\"st\"].tx)\n",
        "plt.figure(figsize=(7, 3.5)); g = tab.groupby(\"supplier\")[\"ratio\"].min()\n",
        "plt.bar([s[-1] for s in g.index], g.values, color=[\"tab:red\" if v < CONF[\"supply_ratio_cut\"] else \"tab:blue\" for v in g.values]); plt.axhline(CONF[\"supply_ratio_cut\"], color=\"k\", ls=\"--\")\n",
        "plt.xlabel(\"supplier\"); plt.ylabel(\"recent / early inbound ratio (min)\"); plt.title(\"Supply disruption scan (seed 17)\"); plt.tight_layout(); plt.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "<details>\n",
        "<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
        "\n",
        "```text\n",
        "[교육용가상자료] 통합 실행 1.2초\n",
        "[설치 스크립트(원클릭) 5단계 점검]\n",
        "  [PASS] 파이썬 버전\n",
        "  [PASS] 가상환경\n",
        "  [PASS] 의존 패키지\n",
        "  [PASS] 자체 시험\n",
        "  [PASS] 데모 실행\n",
        "설치 스크립트 8줄, 단계 5/5 통과 (원클릭 설치부터 실행까지: 달성)\n",
        "\n",
        "[End-to-End 4종: 시드 4개, 업체 1곳 공급 중단, 노드 1곳 3일치 미동기, 변조 20·깨진 줄 5·중복 60건 주입]\n",
        "  seed 7: 성공, 반영 1847건·중복 60건·격리 25/25건, 재고 일치 예, 징후 ['가상업체1'] (주입 가상업체1), 검토 큐 6건, 감사 로그 정상\n",
        "  seed 17: 성공, 반영 1870건·중복 60건·격리 25/25건, 재고 일치 예, 징후 ['가상업체3'] (주입 가상업체3), 검토 큐 6건, 감사 로그 정상\n",
        "  seed 27: 성공, 반영 2030건·중복 60건·격리 25/25건, 재고 일치 예, 징후 ['가상업체5'] (주입 가상업체5), 검토 큐 6건, 감사 로그 정상\n",
        "  seed 37: 성공, 반영 1857건·중복 60건·격리 25/25건, 재고 일치 예, 징후 ['가상업체6'] (주입 가상업체6), 검토 큐 6건, 감사 로그 정상\n",
        "E2E 성공률 100% (목표 100%), 공급 중단 업체 탐지 4/4, 다른 업체 오경보 0건\n",
        "[자체 점검]\n",
        "  [PASS] 정상 자료 통합 실행(오류 주입 포함)\n",
        "  [PASS] 서버 재고 = 반영된 레코드 합산\n",
        "  [PASS] 같은 묶음 재전송은 상태를 바꾸지 않음(멱등)\n",
        "  [PASS] 파싱·범위 오류 줄 격리\n",
        "  [PASS] 충돌 탐지(동시 수정만 충돌)\n",
        "  [PASS] 공급 중단 업체 탐지 + 오프라인 노드로 오경보 없음\n",
        "  [PASS] 정상 자료에서 공급 징후 오경보 0\n",
        "  [PASS] 모든 검토 건이 PENDING(자동 확정 없음)\n",
        "  [PASS] 감사 로그 변조 탐지\n",
        "  [PASS] 대시보드 요약 4개 지표\n",
        "  [PASS] 매뉴얼 7개 장·설치 스크립트 5단계\n",
        "자체 점검 11/11 통과, 함수 커버리지 100%(15/15, 목표 85% 이상: 달성, 미호출 없음)\n",
        "매뉴얼 작성률 7/7장 = 100%: 1. 목적과 범위, 2. 설치와 실행, 3. 입력 데이터, 4. 동기화와 충돌, 5. 공급망 재난 징후, 6. 사람 검토와 감사 로그, 7. 한계와 확장\n",
        "\n",
        "[군수관 브리핑 요약]\n",
        "1. 동기화: 반영 1847건, 중복 거부 60건, 격리 25건. 미동기 177건(1개 보급소).\n",
        "2. 검토 큐 6건(충돌 5, 공급 징후 1): 모두 PENDING.\n",
        "3. 공급망 재난 징후 후보 업체: 가상업체1 (최근 3일 입고량이 첫 3일의 30% 미만).\n",
        "4. 병합 확정과 보급 조치는 군수관·지휘관이 한다. 본 도구는 가상 자료 기반 판단 보조이다.\n",
        "※ 가상 자료·메모리 모의이며 실제 통신·저장 장치·키 관리·인접 체계 연동은 포함하지 않는다. 함수 커버리지는 최상위 함수 호출 기준이다.\n",
        "```\n",
        "</details>\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}