147 KiB
147 KiB
🎯 [DAT06] 월간 정산·보고·취합 One-click 자동화
Google Colab 실습 환경 (데이터·자동화·업무플랫폼 / Level 3)
📌 1. 프로젝트 개요 및 배경
| 항목 | 내용 |
|---|---|
| 과정 분류 | 데이터·자동화·업무플랫폼 (Level 3) |
| 학습 소요 시간 | 10개 미션 (5일 / 총 27.0시간) |
| 환경 사양 | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |
📖 프로젝트 목표
미국 국방재무회계국(DFAS)은 반복적인 재무·정산 업무에 업무 자동화(RPA)를 도입해 사람이 하던 취합과 점검을 줄이는 사업을 추진해 왔다. 이 공개 사례의 학습 초점은 반복 업무 분해, 입력과 트리거, 자동 처리 순서, 오류 복구, 실행 로그, 시간 절감이다. 본 프로젝트는 이를 바탕으로 교육용 가상 부대 24개의 월간 결산 자료(물자 불출, 수리부속 소모, 탄약 정산, 장비 가동률)를 코드로 만들고, 파일 대신 메모리 안의 가상 문서와 표로 모의한다. 학습자는 업무 분해, 다형식 파싱, 검증 룰 엔진, 이상 원인 추천, 일관성 검증, 승인·반려 대시보드, 3단계 사람 결재선, 대량 배치 시험, AAR 보정, 최종 통합을 차례로 수행한다. 최종 목표는 월간 정산·보고·취합을 한 번에 실행하되 발간은 반드시 사람이 결재하도록 하는 원클릭 자동화 도구를 완성하는 것이다.
⚠️ [교육용 가상 데이터셋 원칙]
- 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 코드 내 난수 기반 가상 데이터로 생성됩니다.
- 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.
In [ ]:
# ============================================================================
# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트
# ============================================================================
!pip install -q numpy pandas matplotlib scipy scikit-learn
import os
import sys
import time
import math
import json
import re
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')
# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)
plt.rcParams['font.sans-serif'] = 'DejaVu Sans'
plt.rcParams['axes.unicode_minus'] = False
print(f"실행 환경 준비 완료 (Python {sys.version.split()[0]})")
In [ ]:
"""군수 월간 결산 4개 항목(물자 불출, 수리부속 소모, 탄약 정산, 장비 가동률)의 데이터 구조를 분석해 스키마를 정의하고, 월간 결산 업무를 AS-IS/TO-BE 단계로 분해해 병목 구간을 도출한다."""
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
# 1) 데이터 구조 분석: 원장 4개 항목의 필드·허용값·산식
led = pd.DataFrame(make_ledger(2), columns=LEDGER_COLS)
eq = pd.DataFrame(make_equip(2))
SCHEMA = {
"물자불출": {"fields": "month, unit, item, open, recv, issue, ret, close", "rule": "close = open + recv + ret - issue"},
"수리부속": {"fields": "month, unit, item, open, recv, issue(소모), ret, close", "rule": "close = open + recv + ret - issue"},
"탄약정산": {"fields": "month, unit, item, open, recv, issue(사용), ret(잔탄 반납), close", "rule": "close = open + recv + ret - issue"},
"장비가동률": {"fields": "month, unit, hold(보유), operable(가동)", "rule": "가동률 = operable / hold, 0 <= 가동률 <= 1"},
}
print("[결산 항목별 데이터 구조] [교육용가상자료]")
for k, v in SCHEMA.items():
print(f" - {k}: 필드[{v['fields']}] 규칙[{v['rule']}]")
print(f"\n원장 {len(led)}행(부대 {led['unit'].nunique()}개 x 품목 {led['item'].nunique()}종 x 2개월), 장비 현황 {len(eq)}행")
chk = ((led["open"] + led["recv"] + led["ret"] - led["issue"]) == led["close"]).mean()
cont = led.pivot_table(index=["unit", "item"], columns="month", values=["open", "close"])
cont_ok = (cont[("close", "2026-07")] == cont[("open", "2026-08")]).mean()
print(f"재고 산식 일치율 {chk:.0%}, 월 간 연속성(전월 기말=당월 기초) {cont_ok:.0%}")
print("분류별 월 불출·소모량 합계:")
print(led[led["month"] == "2026-08"].groupby("cat")["issue"].sum().to_string())
print(f"장비 평균 가동률 {(eq['operable'] / eq['hold']).mean():.1%} (가동률 낮은 부대 3개: "
f"{', '.join(eq.assign(r=eq['operable'] / eq['hold']).nsmallest(3, 'r')['unit'])})")
# 2) 월간 결산 업무 AS-IS 분해(단계, 소요 분 - 가상 측정값)와 TO-BE
steps = [("예하 부대별 결산 자료 수신(메일·메신저)", 40, "제출 양식 표준화·자동 수집 함수", 3),
("제출 양식 확인(서식 다른 파일 구분)", 35, "형식 자동 판별·별칭 사전", 2),
("엑셀 값 복사·붙여넣기 취합", 90, "자동 파싱·병합", 2),
("품목 코드·단위 수기 통일", 60, "코드·단위 표준 변환표", 3),
("기초·기말 재고 산식 대조", 55, "검증 룰 15종 자동 적용", 2),
("전월 대비 증감 계산", 30, "전월 비교표 자동 생성", 1),
("이상 수치 확인(전화 문의)", 75, "이상치 후보·원인 요약 추천", 15),
("장비 가동률 집계", 30, "가동률 자동 계산", 1),
("결산 보고서 서식 작성", 80, "템플릿에 자동 채움", 3),
("보급관·군수과장 결재 요청·회수", 45, "3단계 전자 결재 흐름", 15),
("반려 사항 수정 후 재제출", 50, "반려 사유별 수정 위치 안내", 10),
("최종본 배포·보관", 20, "발간본 자동 보관·배포", 1)]
proc = pd.DataFrame(steps, columns=["업무 단계", "AS-IS(분)", "TO-BE 개선", "TO-BE(분)"])
proc.index = range(1, len(proc) + 1)
thr = proc["AS-IS(분)"].mean() + 0.5 * proc["AS-IS(분)"].std()
proc["병목"] = np.where(proc["AS-IS(분)"] >= thr, "병목", "")
print(f"\n[AS-IS / TO-BE 업무 단계 {len(proc)}단계] (병목 기준: 평균+0.5표준편차 = {thr:.0f}분 이상)")
print(proc[["업무 단계", "AS-IS(분)", "TO-BE(분)", "병목"]].to_string())
nb = int((proc["병목"] == "병목").sum())
a, t = proc["AS-IS(분)"].sum(), proc["TO-BE(분)"].sum()
print(f"\n업무 단계 {len(proc)}단계(목표 10단계 이상: {'달성' if len(proc) >= 10 else '미달'}), 병목 {nb}개소(목표 3개소 이상: {'달성' if nb >= 3 else '미달'})")
print(f"1회 결산 AS-IS {a}분 -> TO-BE {t}분 ({1 - t / a:.0%} 단축 예상, 가정값)")
In [ ]:
"""CSV, 엑셀 내보내기(탭 구분), JSON, 공문 텍스트 4가지 형식의 가상 결산 자료를 메모리 안에서 자동 파싱하고 머리글 별칭 대응·정합성 검증·전처리를 거쳐 표준 원장으로 합치는 파이프라인을 구축한다."""
import csv
import io
import json
import re
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
TRUTH = [r for r in make_ledger(2) if r["month"] == "2026-08"]
FIELDS = ["unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
ALIAS = {"unit": ["unit", "부대", "부대명", "제출부대"], "cat": ["cat", "구분", "분류"], "item": ["item", "품목", "품목명"],
"open": ["open", "기초", "기초재고"], "recv": ["recv", "수령", "입고"], "issue": ["issue", "불출", "소모", "사용"],
"ret": ["ret", "반납"], "close": ["close", "기말", "기말재고"]}
H2F = {h: f for f, hs in ALIAS.items() for h in hs}
by_unit = lambda a, b: [r for r in TRUTH if UNITS[a] <= r["unit"] <= UNITS[b]]
# --- 가상 원본 4종(문자열) 생성: 형식마다 서식이 다르다 ---
def mk_csv(rows):
return "\n".join(["부대,구분,품목,기초,수령,불출,반납,기말"] + [",".join(str(r[f]) for f in FIELDS) for r in rows])
def mk_tab(rows): # 엑셀에서 '탭 구분 텍스트'로 내보낸 형태, 머리글이 양식 개정으로 바뀜
return "\n".join(["제출부대\t분류\t품목명\t기초재고\t입고\t소모\t반납\t기말재고"] + ["\t".join(str(r[f]) for f in FIELDS) for r in rows])
def mk_json(rows):
return json.dumps({"form": "결산서", "rows": [{f: r[f] for f in FIELDS} for r in rows]}, ensure_ascii=False)
def mk_memo(rows): # 공문 본문 형태 비정형 텍스트
return "\n".join(f"{i + 1}. {r['unit']} {r['cat']} {r['item']}: 기초 {r['open']}, 수령 {r['recv']}, 불출 {r['issue']}, 반납 {r['ret']}, 기말 {r['close']}"
for i, r in enumerate(rows))
RAW = {"csv": mk_csv(by_unit(0, 2)), "excel_tab": mk_tab(by_unit(3, 5)), "json": mk_json(by_unit(6, 8)), "memo": mk_memo(by_unit(9, 11))}
RAW["csv"] += "\n가상99대대,물자불출,급식류,10,0,3,0,7" # 존재하지 않는 부대 코드(보류 대상)
RAW["memo"] += "\n99. 가상10대대 탄약정산 훈련탄A: 기초 50, 수령 10, 불출 abc, 반납 0, 기말 60" # 숫자 아님(보류 대상)
# --- 파서: 모두 표준 레코드(dict) 리스트로 변환 ---
def norm(row):
out = {}
for k, v in row.items():
f = H2F.get(str(k).strip())
if f:
out[f] = v.strip() if isinstance(v, str) else v
return out
def parse_csv(s, delim=","):
return [norm(r) for r in csv.DictReader(io.StringIO(s), delimiter=delim)]
def parse_json(s):
return [norm(r) for r in json.loads(s)["rows"]]
def parse_memo(s):
pat = re.compile(r"\d+\.\s*(\S+)\s+(\S+)\s+(\S+):\s*기초\s*(\S+),\s*수령\s*(\S+),\s*불출\s*(\S+),\s*반납\s*(\S+),\s*기말\s*(\S+)")
return [dict(zip(FIELDS, m.groups())) for m in map(pat.match, s.splitlines()) if m]
PARSERS = {"csv": parse_csv, "excel_tab": lambda s: parse_csv(s, "\t"), "json": parse_json, "memo": parse_memo}
# --- 전처리·정합성 검증: 형 변환, 코드 확인, 산식 대조. 오류 레코드는 반영 보류 ---
known_items = {it for v in ITEMS.values() for it in v}
def clean(r, fmt):
try:
for f in ("open", "recv", "issue", "ret", "close"):
r[f] = int(r[f])
except (ValueError, KeyError):
return None, f"[{fmt}] 숫자 변환 실패: {r.get('unit')} {r.get('item')}"
if r["unit"] not in UNITS:
return None, f"[{fmt}] 존재하지 않는 부대 코드 {r['unit']}"
if r["item"] not in known_items:
return None, f"[{fmt}] 알 수 없는 품목 {r['item']}"
if r["open"] + r["recv"] + r["ret"] - r["issue"] != r["close"]:
return None, f"[{fmt}] 재고 산식 불일치 {r['unit']} {r['item']}"
return r, None
good, hold, stat = [], [], {}
for fmt, text in RAW.items():
try:
recs = PARSERS[fmt](text)
stat[fmt] = {"parse_ok": True, "n": len(recs)}
except Exception as ex:
recs, stat[fmt] = [], {"parse_ok": False, "n": 0, "err": repr(ex)[:50]}
for r in recs:
c, why = clean(r, fmt)
(good if c else hold).append(c or why)
df = pd.DataFrame(good, columns=FIELDS).sort_values(["unit", "item"]).reset_index(drop=True)
print("[형식별 파싱 결과] [교육용가상자료]")
for fmt, s in stat.items():
print(f" - {fmt:<9} 파싱 {'성공' if s['parse_ok'] else '실패'}, 읽은 레코드 {s['n']}건")
rate = sum(s["parse_ok"] for s in stat.values()) / len(stat)
print(f"파싱 성공률 {rate:.0%}({len(stat)}개 형식), 정상 반영 {len(df)}건, 반영 보류 {len(hold)}건")
for w in hold:
print(" 보류:", w)
truth = pd.DataFrame(by_unit(0, 11), columns=FIELDS).sort_values(["unit", "item"]).reset_index(drop=True)
same = df.equals(truth.astype(df.dtypes.to_dict()))
print(f"원본 대비 값 일치: {'전부 일치' if same else '불일치'} ({len(df)}행 = 12개 부대 x 11품목 = {12 * 11}행)")
print("표준 원장 미리보기(앞 4행):")
print(df.head(4).to_string(index=False))
print(f"\n목표(4개 형식 파싱 성공률 100%, 양식 변경 대응, 오류 레코드 보류): {'달성' if rate == 1 and same and len(hold) == 2 else '미달'}")
print("새 서식 대응은 ALIAS 사전에 머리글 별칭만 추가하면 된다(코드 수정 불필요).")
In [ ]:
"""설정 문자열로 분리한 검증 룰 15종(가상 정산 규정 조항 대응)으로 월간 원장을 검사하고, 표준 정산 규칙에 따른 1차 결산 집계 Baseline을 계산하며 처리 지연시간과 경계 조건 단위 테스트를 확인한다."""
import time
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
# 1) 검증 룰 설정: (룰ID, 가상 조항, 조건식(참이어야 정상), 오류 메시지) - 코드 수정 없이 문자열만 바꿔 룰 변경
RULES = [
("V01", "가상규정 제1조", "unit in UNITS", "존재하지 않는 부대"),
("V02", "가상규정 제2조", "item in UNIT_CODE", "알 수 없는 품목"),
("V03", "가상규정 제3조", "cat in ITEMS and item in ITEMS[cat]", "구분과 품목 불일치"),
("V04", "가상규정 제4조", "open >= 0", "기초 재고 음수"),
("V05", "가상규정 제4조", "recv >= 0", "수령량 음수"),
("V06", "가상규정 제4조", "issue >= 0", "불출량 음수"),
("V07", "가상규정 제4조", "ret >= 0", "반납량 음수"),
("V08", "가상규정 제5조", "close == open + recv + ret - issue", "기말 산식 불일치"),
("V09", "가상규정 제6조", "issue <= open + recv", "가용량 초과 불출"),
("V10", "가상규정 제6조", "ret <= issue", "불출보다 큰 반납"),
("V11", "가상규정 제7조", "close >= 0", "기말 재고 음수"),
("V12", "가상규정 제8조", "len(month) == 7 and month[4] == '-'", "결산월 형식 오류"),
("V13", "가상규정 제9조", "cat != '탄약정산' or issue <= 5000", "탄약 월 사용량 한도 초과"),
("V14", "가상규정 제10조", "isinstance(open, int) and isinstance(close, int)", "수량 자료형 오류"),
("V15", "가상규정 제11조", "issue <= max(open + recv, 1) * 0.9 or issue == 0", "불출 비율 과다(90% 초과)"),
]
COMPILED = [(rid, art, compile(expr, rid, "eval"), msg) for rid, art, expr, msg in RULES]
CTX = {"UNITS": UNITS, "UNIT_CODE": UNIT_CODE, "ITEMS": ITEMS}
def check_row(row):
env = {**CTX, **row}
out = []
for rid, art, code, msg in COMPILED:
try:
ok = eval(code, {"__builtins__": {"len": len, "isinstance": isinstance, "int": int, "max": max}}, env)
except Exception:
ok = False
if not ok:
out.append((rid, art, msg))
return out
# 2) 검증 실행 + 지연시간(행당) 측정
rows = make_ledger(2)
rng = np.random.default_rng(3)
for i in rng.choice(len(rows), 6, replace=False): # 오류 6건 주입(룰이 잡는지 확인)
k = int(rng.integers(0, 3))
rows[i] = {**rows[i], **[{"close": rows[i]["close"] + 7}, {"issue": rows[i]["open"] + rows[i]["recv"] + 50}, {"ret": -3}][k]}
lat, found = [], []
for r in rows:
t0 = time.perf_counter()
res = check_row(r)
lat.append((time.perf_counter() - t0) * 1000)
found += [(r["month"], r["unit"], r["item"], *x) for x in res]
lat = np.array(lat)
print(f"[검증 룰 {len(RULES)}종 적용] 대상 {len(rows)}행, 위반 {len(found)}건 [교육용가상자료]")
cnt = pd.Series([f[3] for f in found]).value_counts()
print("룰별 위반 건수:", {k: int(v) for k, v in cnt.items()})
print("위반 예시(3건):")
for f in found[:3]:
print(" ", f)
print(f"룰 처리 지연: 평균 {lat.mean():.3f}ms, 최대 {lat.max():.3f}ms (목표 50ms 이하: {'달성' if lat.max() <= 50 else '미달'}, 환경에 따라 다름)")
# 3) 1차 결산 집계 Baseline: 부대 x 구분 합계, 구분별 통계
clean = pd.DataFrame([r for r in make_ledger(2) if r["month"] == "2026-08"])
pv = clean.pivot_table(index="unit", columns="cat", values="issue", aggfunc="sum")
print("\n[1차 결산 집계 - 8월 불출·소모량 부대별(상위 4개)]")
print(pv.head(4).to_string())
st = pv.agg(["sum", "mean", "std"]).round(1)
print("[구분별 합계·평균·표준편차]")
print(st.to_string())
print(f"전체 기말 재고 합계 {int(clean['close'].sum())}, 보고 대상 부대 {clean['unit'].nunique()}개")
# 4) 경계 조건 단위 테스트
base = {"month": "2026-08", "unit": UNITS[0], "cat": "물자불출", "item": "유류", "open": 100, "recv": 0, "issue": 0, "ret": 0, "close": 100}
tests = [("정상 행", base, []),
("불출 = 가용량 정확히 일치(V15만 위반)", {**base, "issue": 100, "close": 0}, ["V15"]),
("불출 = 가용량+1 (V09, V11, V15)", {**base, "issue": 101, "close": -1}, ["V09", "V11", "V15"]),
("기말 0 허용", {**base, "issue": 90, "close": 10, "ret": 0}, []),
("반납 = 불출 경계 허용", {**base, "issue": 50, "ret": 50, "close": 100}, []),
("탄약 한도 5000 경계", {**base, "cat": "탄약정산", "item": "훈련탄A", "open": 9000, "issue": 5000, "close": 4000}, []),
("탄약 한도 5001 초과", {**base, "cat": "탄약정산", "item": "훈련탄A", "open": 9000, "issue": 5001, "close": 3999}, ["V13"]),
("결산월 형식 오류", {**base, "month": "202608"}, ["V12"]),
("자료형 오류(문자열 수량)", {**base, "open": "100"}, ["V14"])] # 이 경우 V14 포함 여부만 확인
ok = 0
print("\n[경계 조건 단위 테스트]")
for name, r, exp in tests:
got = sorted(x[0] for x in check_row(r))
p = set(exp) <= set(got) if name.startswith("자료형") else got == exp
ok += p
print(f" - {name}: 기대 {exp} / 결과 {got} -> {'PASS' if p else 'FAIL'}")
print(f"단위 테스트 {ok}/{len(tests)} 통과({ok / len(tests):.0%})")
print(f"목표(룰 15종 이상, 지연 50ms 이하, 단위 테스트 100%): {'달성' if len(RULES) >= 15 and lat.max() <= 50 and ok == len(tests) else '미달'}")
In [ ]:
"""전월 대비 급증한 품목을 로버스트 통계로 자동 감지하고, 훈련·정비 일정 등 맥락 정보와 대조해 이상 원인을 요약·추천하는 모듈을 구현해 감지 정확도와 원인 분류 정확도를 측정한다."""
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
# 1) 가상 2개월 데이터 + 정답(숨은 원인) 생성
led = pd.DataFrame(make_ledger(2))
p7 = led[led["month"] == "2026-07"].set_index(["unit", "item"])
p8 = led[led["month"] == "2026-08"].set_index(["unit", "item"])
d = pd.DataFrame({"cat": p8["cat"], "prev": p7["issue"].clip(lower=1), "cur": p8["issue"]})
rng = np.random.default_rng(4)
n = len(d)
d["train_days"] = rng.integers(0, 5, n) # 해당 부대 월 훈련 일수(맥락)
d["repair_orders"] = rng.integers(0, 4, n) # 정비 지시서 건수(맥락)
d["special"] = 0 # 일회성 특별 지급 승인 여부
d["truth"] = "정상"
d["cur"] = (d["prev"] * rng.uniform(0.7, 1.4, n)).round().astype(int) # 정상 변동(-30%~+40%)
CAUSES = ["훈련 증가", "정비 집중", "입력 자릿수 오류", "일회성 특별 지급", "원인 불명"]
idx = rng.choice(n, 36, replace=False)
for k, i in enumerate(idx):
c = CAUSES[k % 5]
d.iloc[i, d.columns.get_loc("truth")] = c
f = {"훈련 증가": rng.uniform(1.8, 3.0), "정비 집중": rng.uniform(1.8, 3.2), "입력 자릿수 오류": 10.0,
"일회성 특별 지급": rng.uniform(2.0, 4.0), "원인 불명": rng.uniform(1.9, 3.0)}[c]
d.iloc[i, d.columns.get_loc("cur")] = int(d.iloc[i]["prev"] * f)
if c == "훈련 증가":
d.iloc[i, d.columns.get_loc("train_days")] = rng.integers(8, 14)
if c == "정비 집중":
d.iloc[i, d.columns.get_loc("repair_orders")] = rng.integers(7, 12)
if c == "일회성 특별 지급":
d.iloc[i, d.columns.get_loc("special")] = 1
for i in rng.choice(n, 8, replace=False): # 정상이지만 활동이 많았던 곳(오탐 유도)
d.iloc[i, d.columns.get_loc("train_days")] = rng.integers(8, 14)
# 2) 감지: 품목별 로그 증가율의 중앙값·MAD 기반 z 점수
d["ratio"] = d["cur"] / d["prev"]
lr = np.log(d["ratio"])
med = lr.groupby(level="item").transform("median")
mad = (lr - med).abs().groupby(level="item").transform("median").clip(lower=0.05) * 1.4826
d["z"] = (lr - med) / mad
d["flag"] = (d["z"] > 2.0) & (d["ratio"] >= 1.6)
# 3) 원인 요약 추천(규칙 기반, 사람이 확인할 후보를 제시)
def recommend(r):
if abs(r["ratio"] - 10) < 1.2:
return "입력 자릿수 오류"
if r["special"]:
return "일회성 특별 지급"
if r["repair_orders"] >= 6:
return "정비 집중"
if r["train_days"] >= 8:
return "훈련 증가"
return "원인 불명"
ACTION = {"훈련 증가": "훈련 계획 대조 후 정상 처리", "정비 집중": "정비 지시서와 소모 수량 대조", "입력 자릿수 오류": "제출 부대에 수치 재확인 요청",
"일회성 특별 지급": "승인 문서 번호 확인", "원인 불명": "담당자가 부대에 직접 문의"}
d["cause"] = np.where(d["flag"], d.apply(recommend, axis=1), "정상")
# 4) 평가
truth_anom = d["truth"] != "정상"
tp = int((d["flag"] & truth_anom).sum()); fp = int((d["flag"] & ~truth_anom).sum()); fn = int((~d["flag"] & truth_anom).sum())
prec, rec = tp / max(tp + fp, 1), tp / max(tp + fn, 1)
acc = float((d["flag"] == truth_anom).mean())
both = d[d["flag"] & truth_anom]
cause_acc = float((both["cause"] == both["truth"]).mean())
total_acc = float(((d["cause"] == d["truth"])).mean())
print(f"[이상 감지] 대상 {n}건(부대 24 x 품목 11), 실제 급증 {int(truth_anom.sum())}건 [교육용가상자료]")
print(f"감지 {int(d['flag'].sum())}건: 정확도 {acc:.1%}, 정밀도 {prec:.1%}, 재현율 {rec:.1%} (TP {tp}, FP {fp}, FN {fn})")
print(f"원인 분류 정확도(감지된 실제 급증 중) {cause_acc:.1%}, 전체 분류(정상 포함) 정확도 {total_acc:.1%}")
print("\n[원인별 감지·추천 현황]")
print(pd.crosstab(both["truth"], both["cause"]).to_string())
print("\n[이상 원인 요약 추천 예시 - 증가율 상위 5건]")
for (u, it), r in d[d["flag"]].sort_values("ratio", ascending=False).head(5).iterrows():
print(f" - {u} {it}: 전월 {int(r['prev'])} -> 당월 {int(r['cur'])}({r['ratio']:.1f}배), 추정 원인 '{r['cause']}' -> 조치: {ACTION[r['cause']]}")
print("추천은 후보 제시이며 최종 원인 판단과 수치 정정은 담당자가 확인한다.")
print(f"\n목표(추천·분류 정확도 85% 이상): 감지 {'달성' if acc >= 0.85 else '미달'}, 원인 분류 {'달성' if cause_acc >= 0.85 else '미달'}")
In [ ]:
"""결산 결과의 회계적 일관성(재고 산식·월 연속성·합계 교차 대조·가동률 범위)을 검증해 부대별 신뢰도 점수를 산출하고, 입력-분석-검토-결과 4개 화면을 텍스트 화면으로 구현한다."""
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
# ---------- 화면 1: 입력 ----------
led = pd.DataFrame(make_ledger(2))
eq = pd.DataFrame(make_equip(2))
cur = led[led["month"] == "2026-08"].copy()
prev = led[led["month"] == "2026-07"].set_index(["unit", "item"])
rng = np.random.default_rng(5)
faults = [] # 가상 오류 주입 8건
for i in rng.choice(cur.index, 8, replace=False):
kind = ["산식", "연속성", "산식", "연속성"][len(faults) % 4]
if kind == "산식":
cur.loc[i, "close"] += int(rng.integers(5, 40))
else:
cur.loc[i, "open"] += int(rng.integers(5, 40)); cur.loc[i, "close"] = cur.loc[i, "open"] + cur.loc[i, "recv"] + cur.loc[i, "ret"] - cur.loc[i, "issue"]
faults.append((cur.loc[i, "unit"], cur.loc[i, "item"], kind))
eq2 = eq[eq["month"] == "2026-08"].copy()
eq2.iloc[3, eq2.columns.get_loc("operable")] = eq2.iloc[3]["hold"] + 4 # 가동 대수가 보유 대수 초과
SCREEN1 = f"원장 {len(cur)}행(월 2026-08), 장비 현황 {len(eq2)}행, 전월 원장 {len(prev)}행 적재 / 양식 검사 통과"
print("[화면 1. 입력]", SCREEN1)
# ---------- 화면 2: 분석(일관성 검사 4종) ----------
issues = []
for r in cur.itertuples():
if r.open + r.recv + r.ret - r.issue != r.close:
issues.append((r.unit, r.item, "C1 재고 산식 불일치", abs(r.open + r.recv + r.ret - r.issue - r.close)))
if prev.loc[(r.unit, r.item), "close"] != r.open:
issues.append((r.unit, r.item, "C2 전월 기말 != 당월 기초", abs(prev.loc[(r.unit, r.item), "close"] - r.open)))
unit_sum = cur.groupby("unit")[["open", "recv", "ret", "issue", "close"]].sum()
cat_sum = cur.groupby(["unit", "cat"])[["open", "recv", "ret", "issue", "close"]].sum().groupby("unit").sum()
for u in unit_sum.index: # C3 교차 대조: 부대 합계 = 분류 합계의 합
if not (unit_sum.loc[u] == cat_sum.loc[u]).all():
issues.append((u, "-", "C3 합계 교차 대조 불일치", 0))
for r in eq2.itertuples():
if not (0 <= r.operable <= r.hold):
issues.append((r.unit, "장비", "C4 가동 대수가 보유 대수 초과", r.operable - r.hold))
iss = pd.DataFrame(issues, columns=["부대", "품목", "검사", "차이"])
print(f"[화면 2. 분석] 검사 4종 수행, 불일치 {len(iss)}건")
print(iss["검사"].value_counts().to_string())
# ---------- 신뢰도 점수 ----------
score = pd.Series(100.0, index=UNITS)
W = {"C1": 12, "C2": 10, "C3": 15, "C4": 8}
for r in iss.itertuples():
score[r.부대] -= W[r.검사[:2]]
score = score.clip(lower=0)
grade = pd.cut(score, [-1, 69, 89, 100], labels=["C(재제출)", "B(검토)", "A(양호)"])
# ---------- 화면 3: 검토 ----------
print(f"\n[화면 3. 검토] 신뢰도 100점 미만 부대와 검토 대기 사항 [교육용가상자료]")
low = iss.groupby("부대").agg(건수=("검사", "count"), 유형=("검사", lambda s: ", ".join(sorted(set(x[:2] for x in s))))).join(score.rename("점수"))
print(low.sort_values("점수").to_string())
print("검토자는 불일치 건별로 '수정 요청 / 정상 확인' 중 선택하며, 수정은 제출 부대가 한다(시스템이 값을 임의로 바꾸지 않는다).")
# ---------- 화면 4: 결과 ----------
detected = {(u, i) for u, i, k in faults}
caught = {(r.부대, r.품목) for r in iss.itertuples() if r.검사[:2] in ("C1", "C2")}
print(f"\n[화면 4. 결과] 부대 {len(score)}개 중 A {int((grade == 'A(양호)').sum())} / B {int((grade == 'B(검토)').sum())} / C {int((grade == 'C(재제출)').sum())}")
print(f"평균 신뢰도 {score.mean():.1f}점, 주입 오류 {len(faults)}건 중 검출 {len(detected & caught)}건({len(detected & caught) / len(faults):.0%}), 가동률 범위 오류 {int((iss['검사'].str[:2] == 'C4').sum())}건")
ok = len(detected & caught) == len(faults)
print(f"\n목표(핵심 화면 4종 구현, 일관성 검증 엔진 동작): {'달성' if ok else '미달'} (화면 4종: 입력/분석/검토/결과)")
In [ ]:
"""부대별 결산 보고서를 생성하고 제출·승인·반려 상태를 관리하는 대시보드(상태 현황판·그래프)를 구현하며, 입력 검증기와 에러 핸들러로 예외 10종을 크래시 없이 로그에 기록하고 사용자 조치 안내를 출력한다."""
import json
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
LEDGER = pd.DataFrame(make_ledger(2))
LOG, REPORTS = [], {}
class AppError(Exception):
def __init__(self, code, msg, guide):
super().__init__(msg); self.code, self.msg, self.guide = code, msg, guide
def validate(unit, month, rows):
if unit not in UNITS:
raise AppError("E01", f"존재하지 않는 부대 '{unit}'", "부대 목록(가상01~24대대)에서 선택")
if month not in set(LEDGER["month"]):
raise AppError("E02", f"결산월 '{month}' 자료 없음", "2026-07 또는 2026-08 중 선택")
if len(rows) == 0:
raise AppError("E03", "해당 부대 자료가 비어 있음", "제출 부대에 원장 재제출 요청")
for c in ("open", "recv", "issue", "ret", "close"):
if c not in rows.columns:
raise AppError("E04", f"필수 열 '{c}' 누락", "표준 양식을 다시 내려받아 작성")
if not pd.api.types.is_numeric_dtype(rows[c]):
raise AppError("E05", f"'{c}' 열에 숫자가 아닌 값", "숫자만 입력(단위 문자 제거)")
if (rows[c] < 0).any():
raise AppError("E06", f"'{c}' 열에 음수", "수량 0 이상으로 수정")
def generate(unit, month, data=None):
rows = data if data is not None else LEDGER[(LEDGER["unit"] == unit) & (LEDGER["month"] == month)]
validate(unit, month, rows)
rid = f"RPT-{unit[2:4]}-{month[-2:]}"
if rid in REPORTS:
raise AppError("E07", f"이미 생성된 보고서 {rid}", "기존 보고서를 열어 수정·재제출")
REPORTS[rid] = {"unit": unit, "month": month, "status": "작성", "issue": int(rows["issue"].sum()), "close": int(rows["close"].sum()),
"hist": [("작성", "system")]}
return rid
FLOW = {"제출": ("작성", "제출"), "승인": ("제출", "승인"), "반려": ("제출", "반려"), "재작성": ("반려", "작성")}
def act(rid, action, who, reason=""):
if rid not in REPORTS:
raise AppError("E08", f"보고서 {rid} 없음", "보고서 번호를 목록에서 확인")
if action not in FLOW:
raise AppError("E09", f"알 수 없는 처리 '{action}'", "제출/승인/반려/재작성 중 선택")
need, new = FLOW[action]
if REPORTS[rid]["status"] != need:
raise AppError("E10", f"현재 상태 '{REPORTS[rid]['status']}'에서 '{action}' 불가", f"'{need}' 상태에서만 가능")
if action == "반려" and not reason.strip():
raise AppError("E11", "반려 사유 없음", "반려 사유를 한 줄 이상 입력")
REPORTS[rid]["status"] = new
REPORTS[rid]["hist"].append((new, who + (f": {reason}" if reason else "")))
def safe(fn, *a, **k):
"""에러 핸들러: 모든 예외를 잡아 로그에 남기고 사용자에게 알기 쉬운 메시지와 조치를 돌려준다(크래시 없음)."""
try:
r = fn(*a, **k)
LOG.append({"level": "INFO", "code": "OK", "msg": fn.__name__}); return r
except AppError as e:
LOG.append({"level": "ERROR", "code": e.code, "msg": e.msg, "guide": e.guide}); return None
except Exception as e: # 예상 밖 오류도 기록
LOG.append({"level": "ERROR", "code": "E99", "msg": repr(e)[:50], "guide": "운영 담당자에게 로그 전달"}); return None
# 1) 정상 흐름: 12개 부대 보고서 생성 -> 제출 -> 승인 8 / 반려 2 / 대기 2
rids = [safe(generate, u, "2026-08") for u in UNITS[:12]]
for i, rid in enumerate(rids):
safe(act, rid, "제출", f"담당자{i % 3 + 1}")
if i < 8:
safe(act, rid, "승인", "보급관")
elif i < 10:
safe(act, rid, "반려", "보급관", "기말 재고 산식 재확인 요망")
board = pd.DataFrame([{"보고서": k, **{x: v[x] for x in ("unit", "status", "issue", "close")}} for k, v in REPORTS.items()])
print("[결산 보고서 현황판] [교육용가상자료]")
print(board.rename(columns={"unit": "부대", "status": "상태", "issue": "불출합", "close": "기말합"}).to_string(index=False))
print("상태별:", board["status"].value_counts().to_dict())
safe(act, rids[8], "재작성", "담당자")
print("반려 건 재작성 후 상태:", REPORTS[rids[8]]["status"], "/ 이력:", REPORTS[rids[8]]["hist"][-3:])
# 2) 예외 주입 10종
LOG.clear()
bad = LEDGER[(LEDGER["unit"] == UNITS[20]) & (LEDGER["month"] == "2026-08")].copy()
neg = bad.copy(); neg.iloc[0, neg.columns.get_loc("issue")] = -5
txt = bad.copy(); txt["issue"] = txt["issue"].astype(str) + "개"
nocol = bad.drop(columns=["ret"])
cases = [("부대명 오류", generate, ("가상99대대", "2026-08")), ("월 오류", generate, (UNITS[20], "2027-01")),
("빈 자료", generate, (UNITS[20], "2026-08", bad.iloc[0:0])), ("열 누락", generate, (UNITS[20], "2026-08", nocol)),
("숫자 아님", generate, (UNITS[20], "2026-08", txt)), ("음수", generate, (UNITS[20], "2026-08", neg)),
("중복 생성", generate, (UNITS[0], "2026-08")), ("없는 보고서", act, ("RPT-00-00", "승인", "보급관")),
("상태 위반(미제출 승인)", act, (rids[8], "승인", "보급관")), ("반려 사유 누락", act, (rids[11], "반려", "보급관", " "))]
crash = 0
LOG.clear()
for name, fn, args in cases:
try:
safe(fn, *args)
except BaseException:
crash += 1
errs = [l for l in LOG if l["level"] == "ERROR"]
print("\n[예외 주입 결과 - 사용자 안내 메시지]")
for (name, _, _), l in zip(cases, errs):
print(f" {l['code']} {name:<14} {l['msg']:<28} -> 조치: {l['guide']}")
rate = len(errs) / len(cases)
print(f"예외 {len(cases)}종 중 로그 기록 {len(errs)}건(기록률 {rate:.0%}), 크래시 {crash}건")
print("로그 예시:", json.dumps(errs[0], ensure_ascii=False))
print(f"목표(크래시 0건, 에러 로그 기록률 100%): {'달성' if crash == 0 and rate == 1 else '미달'}")
fig, ax = plt.subplots(1, 2, figsize=(11, 4))
cnt = board["status"].map({"승인": "approved", "반려": "rejected", "제출": "pending"}).fillna("draft").value_counts()
ax[0].bar(cnt.index, cnt.values, color="tab:blue"); ax[0].set_title("Report status")
ax[1].barh(board["unit"].str[2:4], board["issue"], color="tab:gray"); ax[1].set_title("Issued quantity by unit (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""결산 보고서 발간 전 담당자 기안, 보급관 검토, 군수과장 최종 승인의 3단계 사람 결재선을 설계하고, 해시 체인 감사 이력의 무결성과 권한 우회 시도 10종의 차단을 검증한다. 자동 발간은 허용하지 않는다."""
import hashlib
import json
import numpy as np
import pandas as pd
# [교육용가상자료] 가상 직책·인물(가명). 실제 전자 결재 체계와 무관한 학습용 모의
USERS = {"kim": "담당자", "lee": "담당자", "park": "보급관", "choi": "군수과장", "han": "열람자"}
LINE = [("기안", "담당자"), ("검토", "보급관"), ("승인", "군수과장")] # 순서 고정 3단계
AUDIT = []
def _h(prev, rec):
return hashlib.sha256((prev + json.dumps(rec, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()[:16]
def log(doc, user, action, result, note=""):
rec = {"seq": len(AUDIT) + 1, "doc": doc, "user": user, "role": USERS.get(user, "미등록"), "action": action, "result": result, "note": note}
rec["hash"] = _h(AUDIT[-1]["hash"] if AUDIT else "GENESIS", {k: v for k, v in rec.items() if k != "hash"})
AUDIT.append(rec)
def verify_chain():
prev, ok = "GENESIS", 0
for r in AUDIT:
body = {k: v for k, v in r.items() if k != "hash"}
ok += (r["hash"] == _h(prev, body)); prev = r["hash"]
return ok, len(AUDIT)
DOCS = {}
def step(doc, user, action, comment=""):
"""결재 처리. 권한·순서·본인 결재 금지·근거 기재를 모두 통과해야 기록되고, 실패는 차단 기록을 남긴다."""
d = DOCS.setdefault(doc, {"stage": 0, "signers": [], "published": False})
role = USERS.get(user)
if role is None:
return log(doc, user, action, "차단", "미등록 사용자"), False
if d["stage"] >= len(LINE):
return log(doc, user, action, "차단", "이미 결재 완료"), False
need_act, need_role = LINE[d["stage"]]
if action != need_act:
return log(doc, user, action, "차단", f"순서 위반: 현재 단계 '{need_act}'"), False
if role != need_role:
return log(doc, user, action, "차단", f"권한 없음: 역할 '{role}'은(는) '{action}' 불가({need_role} 필요)"), False
if user in d["signers"]:
return log(doc, user, action, "차단", "본인이 앞 단계에 서명한 문서는 다음 단계를 결재할 수 없음"), False
if action != "기안" and len(comment.strip()) < 5:
return log(doc, user, action, "차단", "검토 근거(5자 이상) 미기재"), False
d["stage"] += 1; d["signers"].append(user)
log(doc, user, action, "승인", comment or "기안 상신")
return None, True
def publish(doc, user):
d = DOCS.get(doc)
if not d or d["stage"] < len(LINE):
log(doc, user, "발간", "차단", "3단계 결재 미완료(자동 발간 금지)"); return False
if USERS.get(user) != "담당자":
log(doc, user, "발간", "차단", "발간은 담당자가 결재 완료 후 수동 실행"); return False
d["published"] = True; log(doc, user, "발간", "승인", "결재 3단계 완료 후 수동 발간"); return True
# 1) 정상 결재 흐름 2건
for doc, drafter in (("RPT-2026-08-A", "kim"), ("RPT-2026-08-B", "lee")):
step(doc, drafter, "기안")
step(doc, "park", "검토", "재고 산식 및 전월 연속성 확인 완료")
step(doc, "choi", "승인", "보급관 검토 의견 확인, 발간 승인")
publish(doc, drafter)
# 2) 권한 우회 시도 10종 - 결재가 진행되는 순서에 맞춰 시도하고, 모두 차단되어야 한다
blocked, total = 0, 0
print("[권한 우회 시도 결과] [교육용가상자료]")
def attempt(name, fn):
global blocked, total
n = len(AUDIT); fn(); r = AUDIT[-1]
ok = len(AUDIT) == n + 1 and r["result"] == "차단"
blocked += ok; total += 1
print(f" - {name}: {r['result']} ({r['note']})")
attempt("미등록 사용자 기안", lambda: step("RPT-ATT", "hacker", "기안"))
attempt("열람자가 기안", lambda: step("RPT-ATT", "han", "기안"))
attempt("기안 전에 군수과장이 승인(단계 건너뛰기)", lambda: step("RPT-ATT", "choi", "승인", "바로 승인합니다 확인"))
step("RPT-ATT", "kim", "기안")
attempt("기안자 본인이 검토까지 처리", lambda: step("RPT-ATT", "kim", "검토", "본인이 검토합니다 확인"))
attempt("다른 담당자가 검토 단계 처리", lambda: step("RPT-ATT", "lee", "검토", "다른 담당자 검토 시도 확인"))
attempt("검토 근거 없이 보급관 검토", lambda: step("RPT-ATT", "park", "검토", ""))
attempt("보급관 검토 전에 군수과장이 승인", lambda: step("RPT-ATT", "choi", "승인", "보급관 단계를 생략한 승인 확인"))
attempt("결재 미완료 상태에서 발간", lambda: publish("RPT-ATT", "kim"))
step("RPT-ATT", "park", "검토", "산식 검증 결과 이상 없음 확인")
step("RPT-ATT", "choi", "승인", "최종 승인합니다 확인")
attempt("완결된 문서 재승인 시도", lambda: step("RPT-ATT", "choi", "승인", "한 번 더 승인 시도 확인"))
attempt("보급관이 발간 시도", lambda: publish("RPT-ATT", "park"))
print(f"우회 시도 {total}건 중 차단 {blocked}건({blocked / total:.0%})")
# 3) 이력 무결성: 정상 체인 검증 후 위·변조 시험
ok, tot = verify_chain()
print(f"\n이력 {tot}건 해시 체인 검증: {ok}/{tot} 일치({ok / tot:.0%})")
k = next(i for i, r in enumerate(AUDIT) if r["result"] == "차단")
saved = AUDIT[k]["result"]
AUDIT[k]["result"] = "승인" # 차단 기록을 승인으로 위조
ok2, _ = verify_chain()
AUDIT[k]["result"] = saved
print(f"위조 시험: {k + 1}번 기록 변조 시 검증 {ok2}/{tot} -> 변조 탐지 {'성공' if ok2 < tot else '실패'}, 원복 후 {verify_chain()[0]}/{tot}")
st = pd.DataFrame(AUDIT).groupby("doc")["result"].value_counts().unstack(fill_value=0)
print("\n[문서별 결재 이력 요약]")
print(st.to_string())
pubs = [k for k, v in DOCS.items() if v["published"]]
print("발간 완료 문서:", pubs, "/ 우회 시험 문서 RPT-ATT는 발간되지 않음 / 자동 발간 0건(발간은 담당자 수동 실행)")
print(f"목표(이력 무결성 100%, 우회 차단율 100%): {'달성' if ok == tot and blocked == total and ok2 < tot else '미달'}")
print("실제 결재와 보안 규정은 부대 지침과 지휘계통에 따른다. 본 코드는 학습용 모의이다.")
In [ ]:
"""40개 가상 예하 부대 x 12개월(480건) 월간 결산 보고서를 한 번에 생성하는 배치를 실행하며, 일시 장애와 손상 자료를 주입해 재시도·격리 처리를 확인하고 전체 실행 시간, 자동화 성공률, 수작업 대비 절감률을 측정한다."""
import time
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
UNITS = [f"가상{i:02d}대대" for i in range(1, 41)] # 배치용으로 40개 부대로 확장
t_all = time.perf_counter()
rows = make_ledger(12, seed=8)
led = pd.DataFrame(rows)
lab = {f"2026-{i + 7:02d}": f"{2026 + (i + 6) // 12}-{(i + 6) % 12 + 1:02d}" for i in range(12)} # 2026-07 ~ 2027-06
led["month"] = led["month"].map(lab)
months = sorted(led["month"].unique())
print(f"[배치 입력] 부대 {len(UNITS)}개 x {len(months)}개월 = {len(UNITS) * len(months)}건, 원장 {len(led)}행 [교육용가상자료]")
# 장애 주입: 일시 장애(재시도로 복구) 9건, 손상 자료(격리) 3건
rng = np.random.default_rng(80)
keys = [(u, m) for u in UNITS for m in months]
pick = rng.choice(len(keys), 12, replace=False)
transient = {keys[i] for i in pick[:9]}
corrupt = {keys[i] for i in pick[9:]}
for (u, m) in corrupt: # 기말 수치를 훼손(산식 불일치)
i = led[(led["unit"] == u) & (led["month"] == m)].index[0]
led.loc[i, "close"] = led.loc[i, "close"] + 999
attempts_log = {}
def build_report(u, m, g):
k = (u, m)
attempts_log[k] = attempts_log.get(k, 0) + 1
if k in transient and attempts_log[k] == 1:
raise TimeoutError("저장소 응답 지연(일시 장애)")
bad = (g["open"] + g["recv"] + g["ret"] - g["issue"] != g["close"]).sum()
if bad:
raise ValueError(f"재고 산식 불일치 {int(bad)}건")
s = g.groupby("cat")[["issue", "close"]].sum()
return f"{u} {m} 불출 {int(s['issue'].sum())} 기말 {int(s['close'].sum())} (항목 {len(g)})"
done, quarantine, retries = {}, [], 0
t0 = time.perf_counter()
for (u, m), g in led.groupby(["unit", "month"]):
for attempt in range(1, 4): # 최대 3회 재시도
try:
done[(u, m)] = build_report(u, m, g)
break
except TimeoutError:
retries += 1
except ValueError as ex: # 자료 오류는 재시도해도 소용없으므로 즉시 격리하고 다음 건으로 진행
quarantine.append((u, m, str(ex)))
break
run_s = time.perf_counter() - t0
total = len(keys)
succ = len(done) / total
print(f"\n[배치 실행] 생성 {len(done)}건, 격리 {len(quarantine)}건, 재시도 {retries}회")
for q in quarantine:
print(" 격리(담당자 확인 필요):", q)
print(f"보고서 생성 순수 실행 시간 {run_s:.2f}초, 전체(자료 생성 포함) {time.perf_counter() - t_all:.2f}초 (목표 5분 이내: {'달성' if time.perf_counter() - t_all < 300 else '미달'})")
print(f"자동화 성공률 {succ:.1%} (일시 장애 {len(transient)}건은 재시도로 자동 복구, 손상 자료는 격리 후 보고)")
# 수작업 대비 절감(가정값): 부대 보고서 1건 수작업 25분(미션 1 AS-IS 610분/24개 부대 기준), 자동화 후 사람 검토 2분/건, 격리 건 수동 처리 15분
manual = total * 25
auto = total * 2 + len(quarantine) * 15 + run_s / 60
print(f"\n[업무 소요 비교(가정값)] 수작업 {manual}분({manual / 60:.0f}시간) vs 자동화 {auto:.0f}분({auto / 60:.1f}시간) -> 절감 {1 - auto / manual:.1%}")
# 스트레스: 부하 3배(1,440건) 반복 실행의 처리량
big = list(led.groupby(["unit", "month"]))
t0 = time.perf_counter(); n = 0
for _ in range(3):
for (u, m), g in big:
if (u, m) in corrupt:
continue
attempts_log[(u, m)] = 9 # 일시 장애 주입 해제(처리량 측정 목적)
build_report(u, m, g); n += 1
dt = time.perf_counter() - t0
print(f"부하 시험 {n}건 처리 {dt:.2f}초 (초당 {n / dt:.0f}건), 오류 0건")
print(f"\n목표(전체 5분 이내, 자동화 성공률 99% 이상, 수작업 대비 80% 이상 절감): "
f"{'달성' if run_s < 300 and succ >= 0.99 and 1 - auto / manual >= 0.8 else '미달'}")
In [ ]:
"""사후검토(AAR) 사례에서 결산 수치 오집계 5유형(중복 행, 단위 혼용, 정정 누락, 박스·발 혼용, 품목 별칭 누락)을 재현해 원인을 분석하고 집계 룰셋 v1을 v2로 보정한 뒤, 재집계 결과와 UAT 체크리스트 점수, 개선 요구 반영 현황, 패치 내역을 출력한다."""
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
# 1) 가상 제출 레코드 생성: 정답(순불출 = 불출 - 반납)과 부대가 실제로 낸 제출 형태(오집계 유발 요인 포함)
led = pd.DataFrame(make_ledger(1))
rng = np.random.default_rng(9)
truth = {(r.unit, r.item): r.issue - r.ret for r in led.itertuples()}
subs, doc = [], 0
def add(u, it, kind, qty, uom, name=None):
global doc
doc += 1
subs.append({"doc": f"D{doc:05d}", "unit": u, "item": name or it, "kind": kind, "qty": qty, "uom": uom})
for r in led.itertuples():
add(r.unit, r.item, "issue", r.issue, UNIT_CODE[r.item]); add(r.unit, r.item, "return", r.ret, UNIT_CODE[r.item])
keys = list(truth)
pick = lambda n, f=lambda k: True: [keys[i] for i in rng.choice([j for j, k in enumerate(keys) if f(k)], n, replace=False)]
CASE = {}
for k in pick(6): # 사례1: 같은 문서번호가 중복 제출
CASE.setdefault("중복 행", []).append(k); subs.append(dict(next(s for s in subs if (s["unit"], s["item"]) == k and s["kind"] == "issue")))
for k in pick(4, lambda k: k[1] == "유류"): # 사례2: 유류를 kL 단위로 제출(1kL = 1000L)
CASE.setdefault("단위 혼용(kL)", []).append(k)
for s in subs:
if (s["unit"], s["item"]) == k:
s["qty"], s["uom"] = s["qty"] / 1000, "kL"
for k in pick(5): # 사례3: 불출 후 정정 문서(음수 조정)를 별도로 제출
CASE.setdefault("정정 누락", []).append(k); d = int(rng.integers(5, 20))
next(s for s in subs if (s["unit"], s["item"]) == k and s["kind"] == "issue")["qty"] += d
add(k[0], k[1], "adjust", -d, UNIT_CODE[k[1]])
for k in pick(4, lambda k: k[1].startswith("훈련탄")): # 사례4: 탄약을 박스(BX, 1BX=20발)로 제출
CASE.setdefault("박스·발 혼용", []).append(k)
for s in subs:
if (s["unit"], s["item"]) == k:
s["qty"], s["uom"] = s["qty"] / 20, "BX"
for k in pick(5, lambda k: k[1] == "피복류"): # 사례5: 품목명을 줄여서 제출('피복')
CASE.setdefault("품목 별칭 누락", []).append(k)
for s in subs:
if (s["unit"], s["item"]) == k:
s["item"] = "피복"
S = pd.DataFrame(subs)
# 2) 집계 룰셋 v1(AAR 이전) / v2(보정 후)
def aggregate(S, ruleset):
d = S.copy()
if ruleset == "v2":
d = d.drop_duplicates("doc")
d["item"] = d["item"].replace({"피복": "피복류"})
d = d[d["item"].isin(UNIT_CODE)]
FACTOR = {("kL", "L"): 1000, ("BX", "RD"): 20} # (제출 단위, 품목 기준 단위) -> 환산 계수
d["qty"] = d["qty"] * [FACTOR.get((u, UNIT_CODE[i]), 1) for u, i in zip(d["uom"], d["item"])]
sign = {"issue": 1, "return": -1, "adjust": 1}
else:
d = d[d["item"].isin(UNIT_CODE) & (d["kind"] != "adjust")]
sign = {"issue": 1, "return": -1}
d = d.assign(v=d["kind"].map(sign) * d["qty"])
return d.groupby(["unit", "item"])["v"].sum().round(6)
def compare(agg):
tt = pd.Series(truth)
tt.index.names = ["unit", "item"]
diff = (agg.reindex(tt.index).fillna(0) - tt)
return diff[diff.abs() > 0.5]
v1, v2 = compare(aggregate(S, "v1")), compare(aggregate(S, "v2"))
print(f"[AAR 오집계 분석] 제출 레코드 {len(S)}건, 정답 대조 대상 {len(truth)}개 (부대 x 품목) [교육용가상자료]")
print(f"룰셋 v1 오집계 {len(v1)}건(오차 합계 {v1.abs().sum():.0f}), 룰셋 v2 오집계 {len(v2)}건")
print("\n[사례별 원인 분석 - v1]")
rows = []
for name, ks in CASE.items():
hit = [k for k in ks if k in v1.index]
fix = [k for k in ks if k not in v2.index]
err = float(sum(abs(v1[k]) for k in hit))
rows.append({"사례": name, "발생 셀": len(ks), "v1 오류 셀": len(hit), "v1 오차": round(err), "v2 정상": f"{len(fix)}/{len(ks)}"})
print(pd.DataFrame(rows).to_string(index=False))
CAUSE = {"중복 행": "문서번호 중복 제거 규칙 없음", "단위 혼용(kL)": "단위 환산표 없음", "정정 누락": "정정(음수 조정) 문서를 집계에서 제외",
"박스·발 혼용": "BX->RD 환산표 없음", "품목 별칭 누락": "별칭 사전 없고 미등록 품목을 조용히 버림"}
print("\n[원인 -> 룰셋 보정]")
for k, v in CAUSE.items():
print(f" - {k}: {v} -> v2 패치: " + {"중복 행": "doc 기준 drop_duplicates", "단위 혼용(kL)": "kL x1000", "정정 누락": "adjust 포함 집계",
"박스·발 혼용": "BX x20", "품목 별칭 누락": "별칭 사전 + 미등록 품목은 로그로 보류"}[k])
# 3) 회귀 시험 및 UAT 체크리스트(자동 점검 20항목)
agg2 = aggregate(S, "v2")
checks = [("정답 대조 오집계 0건", len(v2) == 0), ("중복 문서 제거", agg2.equals(aggregate(S.drop_duplicates("doc"), "v2"))),
("v2 재실행 결과 동일(멱등)", aggregate(S, "v2").equals(agg2)),
("kL 부대 4곳 정상", all(k not in v2.index for k in CASE["단위 혼용(kL)"])), ("BX 부대 4곳 정상", all(k not in v2.index for k in CASE["박스·발 혼용"])),
("정정 문서 5건 반영", all(k not in v2.index for k in CASE["정정 누락"])), ("별칭 5건 반영", all(k not in v2.index for k in CASE["품목 별칭 누락"])),
("정상 부대 v1 결과와 불변", len(set(truth) - set(v1.index) - set(v2.index)) == len(truth) - len(v1)),
("전체 불출 합계 정답과 일치", abs(agg2.sum() - sum(truth.values())) < 1), ("부대 24개 모두 집계", agg2.index.get_level_values(0).nunique() == 24),
("품목 11종 모두 집계", agg2.index.get_level_values(1).nunique() == 11), ("음수 순불출 0건", (agg2 >= 0).all()),
("미등록 품목 입력 시 오류 아닌 보류", aggregate(pd.concat([S, pd.DataFrame([{"doc": "X1", "unit": UNITS[0], "item": "미등록", "kind": "issue", "qty": 5, "uom": "EA"}])]), "v2").shape == agg2.shape),
("빈 입력에서도 오류 없음", aggregate(S.iloc[0:0], "v2").empty), ("처리 결과 부대x품목 단일 키", agg2.index.is_unique),
("v1 대비 오차 감소 100%", len(v2) == 0 and len(v1) > 0), ("단위 환산 후 소수 오차 없음", bool(((agg2 - agg2.round()).abs() < 1e-6).all())),
("사례 5유형 모두 재현·분석", len(CASE) == 5), ("원인-패치 대응표 작성", len(CAUSE) == 5), ("개선 요구 5건 반영", len(CASE) == 5 and len(v2) == 0)]
passed = sum(ok for _, ok in checks)
score = 100 * passed / len(checks)
print("\n[UAT 체크리스트]", f"{passed}/{len(checks)}항목 통과, 점수 {score:.0f}점")
for n, ok in checks:
if not ok:
print(" 미통과:", n)
print("\n[개선 요구 반영 현황]")
for i, (k, ks) in enumerate(CASE.items(), 1):
print(f" 요구 {i}. {k} 대응: {'반영 완료' if all(x not in v2.index for x in ks) else '미반영'}")
print(f"\n목표(UAT 90점 이상, 개선 요구 5건 반영): {'달성' if score >= 90 and len(v2) == 0 else '미달'}")
print("패치 내역: 룰셋 v1 -> v2 (중복 제거, 단위 환산표, 정정 반영, 별칭 사전, 미등록 품목 보류). 재집계 전 담당자가 보류 건을 확인한다.")
In [ ]:
"""환경 점검부터 4형식 자료 수집, 검증 룰 적용, 월간 결산 집계, 이상치 점검, 신뢰도 점수, 24개 부대 보고서 생성, 3단계 결재 대기 분리까지 한 번에 실행하는 원클릭 결산 도구를 완성하고 설치 스크립트 개요·매뉴얼 목차·자체 점검·브리핑 요약을 출력한다."""
import csv
import importlib.util
import io
import json
import sys
import time
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 월간 결산 원장 생성기 (실제 부대명·품목·수량·재무 정보 없음)
# 가상 대대 24개 x (물자 불출 / 수리부속 소모 / 탄약 정산) 품목 원장과 장비 가동 현황을 만든다.
# 재고 산식: 기말 = 기초 + 수령 + 반납 - 불출, 다음 달 기초 = 이번 달 기말.
# ---------------------------------------------------------------------------
UNITS = [f"가상{i:02d}대대" for i in range(1, 25)]
ITEMS = {"물자불출": ["급식류", "피복류", "유류", "정비소모품"],
"수리부속": ["필터류", "베어링류", "전장품", "타이어류"],
"탄약정산": ["훈련탄A", "훈련탄B", "모의탄"]}
UNIT_CODE = {"급식류": "BX", "피복류": "EA", "유류": "L", "정비소모품": "EA", "필터류": "EA", "베어링류": "EA",
"전장품": "EA", "타이어류": "EA", "훈련탄A": "RD", "훈련탄B": "RD", "모의탄": "RD"}
LEDGER_COLS = ["month", "unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
def make_ledger(n_months=2, seed=60):
"""n_months개월 원장(list of dict). 둘째 달 기초는 첫째 달 기말과 같다."""
rng = np.random.default_rng(seed)
rows, prev = [], {}
for m in range(n_months):
for u in UNITS:
for cat, items in ITEMS.items():
for it in items:
base = int(rng.integers(80, 400))
op = prev.get((u, it), base * 3)
recv = int(rng.integers(0, base))
issue = int(rng.integers(base // 4, base // 2 + 1))
ret = int(issue * rng.uniform(0, 0.15))
cl = op + recv + ret - issue
rows.append({"month": f"2026-{m + 7:02d}", "unit": u, "cat": cat, "item": it,
"open": op, "recv": recv, "issue": issue, "ret": ret, "close": cl})
prev[(u, it)] = cl
return rows
def make_equip(n_months=2, seed=61):
rng = np.random.default_rng(seed)
out = []
for m in range(n_months):
for u in UNITS:
hold = int(rng.integers(20, 60))
out.append({"month": f"2026-{m + 7:02d}", "unit": u, "hold": hold, "operable": int(hold * rng.uniform(0.78, 0.97))})
return out
LOG = []
def stage(name):
def deco(fn):
def run(*a):
t0 = time.perf_counter()
try:
r = fn(*a); LOG.append((name, "OK", time.perf_counter() - t0)); return r
except Exception as ex: # 한 단계가 실패해도 도구 전체가 죽지 않고 원인을 기록
LOG.append((name, f"FAIL {repr(ex)[:40]}", time.perf_counter() - t0)); return None
return run
return deco
# ---------------- 0) 설치·환경 점검(원클릭 설치 스크립트 개요) ----------------
INSTALL = ["1. 가상 환경 생성 및 필수 모듈 설치(numpy, pandas, matplotlib)", "2. 환경 점검: 모듈·버전 확인", "3. 샘플 자료로 시험 실행", "4. 실행: run_monthly_close()", "5. 결과 확인: 보고서 24건, 결재 대기 목록"]
mods = ["numpy", "pandas", "matplotlib", "csv", "json"]
env_ok = sum(importlib.util.find_spec(m) is not None for m in mods)
print(f"[0] 환경 점검: Python {sys.version_info.major}.{sys.version_info.minor}, 필수 모듈 {env_ok}/{len(mods)} 확인 [교육용가상자료]")
# ---------------- 1) 수집(CSV + JSON 두 형식 혼합) ----------------
@stage("수집·파싱")
def collect():
rows = [r for r in make_ledger(2) if r["month"] == "2026-08"]
half = len(rows) // 2
head = ["unit", "cat", "item", "open", "recv", "issue", "ret", "close"]
csv_text = "\n".join([",".join(head)] + [",".join(str(r[h]) for h in head) for r in rows[:half]])
json_text = json.dumps([{h: r[h] for h in head} for r in rows[half:]], ensure_ascii=False)
a = list(csv.DictReader(io.StringIO(csv_text))) + json.loads(json_text)
df = pd.DataFrame(a)
for c in ("open", "recv", "issue", "ret", "close"):
df[c] = pd.to_numeric(df[c])
return df
# ---------------- 2) 검증 룰 ----------------
RULES = [("V1 산식", lambda d: d["open"] + d["recv"] + d["ret"] - d["issue"] == d["close"]), ("V2 음수", lambda d: (d[["open", "recv", "issue", "ret", "close"]] >= 0).all(axis=1)),
("V3 가용량", lambda d: d["issue"] <= d["open"] + d["recv"]), ("V4 반납", lambda d: d["ret"] <= d["issue"]), ("V5 부대", lambda d: d["unit"].isin(UNITS)),
("V6 품목", lambda d: d["item"].isin(UNIT_CODE))]
@stage("검증")
def check(df):
bad = pd.Series(False, index=df.index)
for _, f in RULES:
bad |= ~f(df)
return int(bad.sum())
@stage("집계")
def aggregate(df):
return df.groupby(["unit", "cat"])[["issue", "ret", "close"]].sum().unstack("cat")
@stage("이상 점검")
def anomalies(df):
prev = pd.DataFrame([r for r in make_ledger(2) if r["month"] == "2026-07"]).set_index(["unit", "item"])["issue"]
ratio = df.set_index(["unit", "item"])["issue"] / prev.clip(lower=1)
return ratio[ratio >= 4.0]
@stage("신뢰도")
def reliability(df, nbad):
return round(100 - 100 * nbad / len(df), 1)
@stage("보고서 생성")
def build(df, agg):
out = {}
for u in UNITS:
g = df[df["unit"] == u]
out[u] = {"unit": u, "issue": int(g["issue"].sum()), "close": int(g["close"].sum()), "status": "결재 대기(기안)", "published": False}
return out
df = collect()
nbad = check(df)
agg = aggregate(df)
anom = anomalies(df)
rel = reliability(df, nbad)
reports = build(df, agg)
# ---------------- 3) 결과 ----------------
print(f"[1] 수집 {len(df)}행(CSV+JSON 혼합), 검증 위반 {nbad}건, 이상 급증 후보 {len(anom)}건, 자료 신뢰도 {rel}점")
tbl = pd.DataFrame(reports).T[["issue", "close", "status"]].head(5)
print("[2] 부대별 결산 보고서(상위 5개 부대) - 모든 보고서는 결재 대기 상태")
print(tbl.to_string())
tot = agg["issue"].sum()
print("[3] 분류별 불출·소모 합계:", tot.astype(int).to_dict())
print("[4] 단계별 실행 로그:")
for n, st, t in LOG:
print(f" - {n:<8} {st:<5} {t * 1000:.1f}ms")
run_ms = sum(t for _, _, t in LOG) * 1000
# ---------------- 4) 자체 점검 ----------------
MANUAL = ["1. 개요·설치", "2. 입력 자료 양식", "3. 실행 방법(원클릭)", "4. 검증 룰 목록", "5. 이상 항목 처리", "6. 결재선·권한", "7. 오류 메시지 조치", "8. 배치 운영", "9. 개선 이력·문의"]
done_sec = len(MANUAL)
checks = [("환경 점검 5/5", env_ok == len(mods)), ("모든 단계 OK", all(s == "OK" for _, s, _ in LOG)), ("24개 부대 보고서 생성", len(reports) == 24),
("자동 발간 0건", not any(r["published"] for r in reports.values())), ("원장 산식 위반 0건", nbad == 0),
("실행 시간 1분 이내", run_ms < 60000), ("매뉴얼 9장 작성(작성률 100%)", done_sec == len(MANUAL)),
("설치 절차 5단계 완비", len(INSTALL) == 5)]
print("\n[5] 자체 점검")
for n, ok in checks:
print(f" {'PASS' if ok else 'FAIL'} {n}")
npass = sum(ok for _, ok in checks)
print("설치 스크립트 개요:", " / ".join(INSTALL))
print("매뉴얼 목차:", " | ".join(MANUAL))
asis, tobe = 610, 58
print(f"\n[브리핑 요약] 24개 부대 결산 보고서를 {run_ms:.0f}ms에 자동 생성(수작업 가정 {asis}분 -> 검토 포함 {tobe}분, {1 - tobe / asis:.0%} 절감 가정값).")
print(f"자료 신뢰도 {rel}점, 이상 급증 후보 {len(anom)}건은 담당자 확인 대상, 보고서 24건 전부 보급관 -> 군수과장 결재 대기(자동 발간 없음).")
print(f"자체 점검 {npass}/{len(checks)} PASS. 목표(원클릭 설치·실행 완비, 매뉴얼 100%): {'달성' if npass == len(checks) else '미달'}")