2520 lines
158 KiB
Markdown
2520 lines
158 KiB
Markdown
# MNT02 정답 코드 - AI 기반 지능형 부품 단종위험 분석·대응체계
|
|
|
|
> [교육용가상자료] 모든 데이터는 코드 안에서 난수로 만든 가상 자료이며 실제 부대·장비·인원·작전 정보와 무관합니다.
|
|
|
|
## 실행 환경
|
|
|
|
- Python 3.10 이상, numpy, pandas, scipy, scikit-learn, matplotlib
|
|
- 미션별 코드는 각각 한 파일로 독립 실행됩니다. 외부 파일 없이 코드 안에서 가상 데이터를 생성합니다.
|
|
- 그래프의 글자는 한글 글꼴이 없는 실행 환경에서도 깨지지 않도록 영문으로 표기했습니다. 표·출력 결과는 한글입니다.
|
|
- 난수 시드를 고정해 두어 같은 코드를 실행하면 같은 결과가 나옵니다.
|
|
|
|
## 미션 1. 야전 정비창 수리부속 청구/출고 이력 및 장비 가동률 데이터셋 스키마 정의
|
|
|
|
- 난이도: 초급 / 예상 소요시간: 1.5시간
|
|
- 코드 설명: 야전 정비창 수리부속 청구/출고 이력과 장비 가동률 데이터의 업무 프로세스 맵, 데이터 스키마(피처 20개 이상)와 데이터 딕셔너리를 정의하고 가상 청구 기록을 검증한다.
|
|
|
|
```python
|
|
"""야전 정비창 수리부속 청구/출고 이력과 장비 가동률 데이터의 업무 프로세스 맵, 데이터 스키마(피처 20개 이상)와 데이터 딕셔너리를 정의하고 가상 청구 기록을 검증한다."""
|
|
import json
|
|
import re
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
# 1) 업무 프로세스 맵: 단계별로 어떤 데이터가 생기는지 연결한다
|
|
PROCESS = [("1. 정비 소요 발생", "장비 고장·계획정비", "equip_id, equip_type, maint_type"),
|
|
("2. 수리부속 청구", "정비반 -> 정비창", "req_id, req_date, nsn, qty_req, priority"),
|
|
("3. 재고 확인·불출", "정비창 재고 담당", "qty_issued, emergency_flag"),
|
|
("4. 조달 요청", "재고 부족 시 조달부서", "lead_time_m, supplier_count, unit_price"),
|
|
("5. 단종 위험 검토", "품목 관리 담당", "prod_status, alt_available"),
|
|
("6. 정비 완료·가동률 집계", "월 단위 집계", "op_hours, availability, downtime_h")]
|
|
print("분석 목표: 월별 품목 수요를 예측해 결품·과잉재고를 줄이고, 단종예고·단종 품목의 최종 구매 수량 판단을 지원한다.")
|
|
print("[업무 프로세스 맵] 정비 소요 -> 청구 -> 불출 -> 조달 -> 단종 검토 -> 가동률")
|
|
for step, who, fields in PROCESS:
|
|
print(f" {step:18s} | {who:18s} | {fields}")
|
|
|
|
# 2) 스키마(데이터 딕셔너리): 그룹, 자료형, 규칙, 설명
|
|
NSN_RE = r"^\d{4}-\d{2}-\d{3}-\d{4}$" # NSN = FSC(4) + 국가코드(2) + 일련번호(7), NIIN = 뒤 9자리
|
|
SCHEMA = {
|
|
"청구/불출": {"req_id": (str, {"pattern": r"^R\d{6}$"}, "청구 번호"),
|
|
"req_date": (str, {"pattern": r"^\d{4}-\d{2}-\d{2}$"}, "청구 일자"),
|
|
"unit_code": (str, {"pattern": r"^U-\d{2}$"}, "청구 부대(가상 코드)"),
|
|
"equip_id": (str, {"pattern": r"^[ABC]-\d{3}$"}, "장비 번호(가상)"),
|
|
"equip_type": (str, {"enum": ["A", "B", "C"]}, "장비 유형"),
|
|
"maint_type": (str, {"enum": ["계획", "비계획"]}, "정비 구분"),
|
|
"nsn": (str, {"pattern": NSN_RE}, "국가재고번호(가상)"),
|
|
"fsc": (str, {"pattern": r"^\d{4}$"}, "군수품 분류(NSN 앞 4자리)"),
|
|
"niin": (str, {"pattern": r"^\d{9}$"}, "품목 식별번호(NSN 뒤 9자리)"),
|
|
"qty_req": (int, {"min": 1, "max": 500}, "청구 수량"),
|
|
"qty_issued": (int, {"min": 0, "max": 500}, "불출 수량"),
|
|
"unit_of_issue": (str, {"enum": ["EA", "SE", "BX"]}, "불출 단위"),
|
|
"priority": (str, {"enum": ["긴급", "일반"]}, "청구 우선순위"),
|
|
"emergency_flag": (bool, {}, "긴급 불출 여부(이상치로 지우지 않음)")},
|
|
"품목/공급망": {"unit_price": (int, {"min": 0}, "단가(원, 가상)"),
|
|
"lead_time_m": (int, {"min": 0, "max": 36}, "조달 리드타임(월)"),
|
|
"supplier_count": (int, {"min": 0, "max": 20}, "공급 가능 업체 수"),
|
|
"prod_status": (str, {"enum": ["생산중", "단종예고", "단종"]}, "생산(단종) 상태"),
|
|
"alt_available": (bool, {}, "대체품 존재 여부")},
|
|
"가동률(월)": {"ym": (str, {"pattern": r"^\d{4}-\d{2}$"}, "집계 월"),
|
|
"op_hours": (float, {"min": 0, "max": 744}, "장비 운용시간"),
|
|
"availability": (float, {"min": 0, "max": 1}, "장비 가동률"),
|
|
"downtime_h": (float, {"min": 0, "max": 744}, "정비 대기·수리 시간"),
|
|
"exercise_flag": (int, {"enum": [0, 1]}, "대규모 훈련 월 여부")},
|
|
}
|
|
|
|
|
|
def check(name, v, typ, rule):
|
|
if typ is float and isinstance(v, int) and not isinstance(v, bool):
|
|
v = float(v)
|
|
if not isinstance(v, typ) or (typ is not bool and isinstance(v, bool)):
|
|
return [f"{name}: 자료형 오류({type(v).__name__})"]
|
|
errs = []
|
|
if "enum" in rule and v not in rule["enum"]:
|
|
errs.append(f"{name}: 허용값 아님 '{v}'")
|
|
if "min" in rule and v < rule["min"]:
|
|
errs.append(f"{name}: 최소값 {rule['min']} 미만")
|
|
if "max" in rule and v > rule["max"]:
|
|
errs.append(f"{name}: 최대값 {rule['max']} 초과")
|
|
if "pattern" in rule and not re.match(rule["pattern"], v):
|
|
errs.append(f"{name}: 형식 불일치 '{v}'")
|
|
return errs
|
|
|
|
|
|
def validate(rec):
|
|
errs = []
|
|
for grp in SCHEMA.values():
|
|
for k, (typ, rule, _) in grp.items():
|
|
errs += [f"{k}: 필수 필드 누락"] if k not in rec else check(k, rec[k], typ, rule)
|
|
if not errs: # 필드 간 논리 규칙
|
|
if rec["nsn"][:4] != rec["fsc"] or rec["nsn"].replace("-", "")[4:] != rec["niin"]:
|
|
errs.append("nsn/fsc/niin 불일치")
|
|
if rec["qty_issued"] > rec["qty_req"]:
|
|
errs.append("불출 수량이 청구 수량 초과")
|
|
return errs
|
|
|
|
|
|
# 3) [교육용가상자료] 청구 기록 20건 생성
|
|
rng = np.random.default_rng(1)
|
|
records = []
|
|
for i in range(20):
|
|
fsc = str(rng.choice(["2520", "2530", "2940", "6140"]))
|
|
niin = f"00{rng.integers(1000000, 9999999)}"
|
|
q = int(rng.integers(1, 20))
|
|
et = str(rng.choice(["A", "B", "C"]))
|
|
records.append({"req_id": f"R{250001 + i}", "req_date": f"2025-{rng.integers(1, 13):02d}-{rng.integers(1, 29):02d}",
|
|
"unit_code": f"U-{rng.integers(1, 6):02d}", "equip_id": f"{et}-{rng.integers(1, 200):03d}", "equip_type": et,
|
|
"maint_type": str(rng.choice(["계획", "비계획"])), "nsn": f"{fsc}-{niin[:2]}-{niin[2:5]}-{niin[5:]}",
|
|
"fsc": fsc, "niin": niin, "qty_req": q, "qty_issued": int(rng.integers(0, q + 1)),
|
|
"unit_of_issue": "EA", "priority": str(rng.choice(["긴급", "일반"], p=[0.1, 0.9])),
|
|
"emergency_flag": bool(rng.random() < 0.05), "unit_price": int(rng.integers(10, 900)) * 1000,
|
|
"lead_time_m": int(rng.integers(1, 13)), "supplier_count": int(rng.integers(1, 5)),
|
|
"prod_status": str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1])),
|
|
"alt_available": bool(rng.random() < 0.5), "ym": "2025-12", "op_hours": float(round(rng.uniform(80, 180), 1)),
|
|
"availability": float(round(rng.uniform(0.75, 0.97), 3)), "downtime_h": float(round(rng.uniform(5, 60), 1)),
|
|
"exercise_flag": int(rng.integers(0, 2))})
|
|
|
|
ok = sum(1 for r in records if not validate(r))
|
|
n_feat = sum(len(g) for g in SCHEMA.values())
|
|
print(f"\n정의한 피처(필드) {n_feat}개 (목표 20개 이상: {'달성' if n_feat >= 20 else '미달'})")
|
|
print(f"정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건")
|
|
|
|
# 4) 오류 주입 시험
|
|
bad = json.loads(json.dumps(records[0], ensure_ascii=False))
|
|
bad["nsn"] = "2520-001234567" # NSN 형식 오류
|
|
bad["qty_req"] = -3 # 음수 수량
|
|
bad["prod_status"] = "미상" # 허용값 아님
|
|
del bad["lead_time_m"] # 필수 필드 누락
|
|
bad["availability"] = 1.4 # 가동률 범위 초과
|
|
print("[오류 주입 결과]")
|
|
for e in validate(bad):
|
|
print(" -", e)
|
|
bad2 = json.loads(json.dumps(records[1], ensure_ascii=False)); bad2["fsc"] = "9999"
|
|
print(" - (논리 규칙)", validate(bad2))
|
|
|
|
# 5) 데이터 딕셔너리 출력
|
|
rows = [(g, k, t.__name__, json.dumps(r, ensure_ascii=False) if r else "-", d) for g, grp in SCHEMA.items() for k, (t, r, d) in grp.items()]
|
|
dic = pd.DataFrame(rows, columns=["그룹", "필드", "자료형", "규칙", "설명"])
|
|
pd.set_option("display.width", 200); pd.set_option("display.max_colwidth", 34)
|
|
print("[데이터 딕셔너리]")
|
|
print(dic.to_string(index=False))
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
분석 목표: 월별 품목 수요를 예측해 결품·과잉재고를 줄이고, 단종예고·단종 품목의 최종 구매 수량 판단을 지원한다.
|
|
[업무 프로세스 맵] 정비 소요 -> 청구 -> 불출 -> 조달 -> 단종 검토 -> 가동률
|
|
1. 정비 소요 발생 | 장비 고장·계획정비 | equip_id, equip_type, maint_type
|
|
2. 수리부속 청구 | 정비반 -> 정비창 | req_id, req_date, nsn, qty_req, priority
|
|
3. 재고 확인·불출 | 정비창 재고 담당 | qty_issued, emergency_flag
|
|
4. 조달 요청 | 재고 부족 시 조달부서 | lead_time_m, supplier_count, unit_price
|
|
5. 단종 위험 검토 | 품목 관리 담당 | prod_status, alt_available
|
|
6. 정비 완료·가동률 집계 | 월 단위 집계 | op_hours, availability, downtime_h
|
|
|
|
정의한 피처(필드) 24개 (목표 20개 이상: 달성)
|
|
정상 샘플 20건 중 검증 통과 20건, 오류 0건
|
|
[오류 주입 결과]
|
|
- nsn: 형식 불일치 '2520-001234567'
|
|
- qty_req: 최소값 1 미만
|
|
- lead_time_m: 필수 필드 누락
|
|
- prod_status: 허용값 아님 '미상'
|
|
- availability: 최대값 1 초과
|
|
- (논리 규칙) ['nsn/fsc/niin 불일치']
|
|
[데이터 딕셔너리]
|
|
그룹 필드 자료형 규칙 설명
|
|
청구/불출 req_id str {"pattern": "^R\\d{6}$"} 청구 번호
|
|
청구/불출 req_date str {"pattern": "^\\d{4}-\\d{2}-\\d{2}$"} 청구 일자
|
|
청구/불출 unit_code str {"pattern": "^U-\\d{2}$"} 청구 부대(가상 코드)
|
|
청구/불출 equip_id str {"pattern": "^[ABC]-\\d{3}$"} 장비 번호(가상)
|
|
청구/불출 equip_type str {"enum": ["A", "B", "C"]} 장비 유형
|
|
청구/불출 maint_type str {"enum": ["계획", "비계획"]} 정비 구분
|
|
청구/불출 nsn str {"pattern": "^\\d{4}-\\d{2}-\\d{3}-\\d{4}$"} 국가재고번호(가상)
|
|
청구/불출 fsc str {"pattern": "^\\d{4}$"} 군수품 분류(NSN 앞 4자리)
|
|
청구/불출 niin str {"pattern": "^\\d{9}$"} 품목 식별번호(NSN 뒤 9자리)
|
|
청구/불출 qty_req int {"min": 1, "max": 500} 청구 수량
|
|
청구/불출 qty_issued int {"min": 0, "max": 500} 불출 수량
|
|
청구/불출 unit_of_issue str {"enum": ["EA", "SE", "BX"]} 불출 단위
|
|
청구/불출 priority str {"enum": ["긴급", "일반"]} 청구 우선순위
|
|
청구/불출 emergency_flag bool - 긴급 불출 여부(이상치로 지우지 않음)
|
|
품목/공급망 unit_price int {"min": 0} 단가(원, 가상)
|
|
품목/공급망 lead_time_m int {"min": 0, "max": 36} 조달 리드타임(월)
|
|
품목/공급망 supplier_count int {"min": 0, "max": 20} 공급 가능 업체 수
|
|
품목/공급망 prod_status str {"enum": ["생산중", "단종예고", "단종"]} 생산(단종) 상태
|
|
품목/공급망 alt_available bool - 대체품 존재 여부
|
|
가동률(월) ym str {"pattern": "^\\d{4}-\\d{2}$"} 집계 월
|
|
... (이하 생략)
|
|
```
|
|
|
|
## 미션 2. 결측치 보정, 계절성 수요 분리 및 수리부속 품목코드(NSN) 정제 전처리 모듈 구현
|
|
|
|
- 난이도: 초급~중급 / 예상 소요시간: 2.0시간
|
|
- 코드 설명: 수리부속 청구/출고 원시 기록에 정제 규칙 10종(NSN 형식 정규화, 중복 제거, 날짜 통일, 단위 환산, 결측 월 계절성 보간 등)을 적용하고, 월별 수요에서 계절성 성분을 분리한 뒤 정합성 검증 통과율을 확인한다.
|
|
|
|
```python
|
|
"""수리부속 청구/출고 원시 기록에 정제 규칙 10종(NSN 형식 정규화, 중복 제거, 날짜 통일, 단위 환산, 결측 월 계절성 보간 등)을 적용하고, 월별 수요에서 계절성 성분을 분리한 뒤 정합성 검증 통과율을 확인한다."""
|
|
import re
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
master = set(items["nsn"])
|
|
nsn_of = dict(zip(items["item_id"], items["nsn"]))
|
|
|
|
# 1) [교육용가상자료] 최근 24개월 월 수요를 개별 청구 기록으로 풀고 일부러 오류를 섞는다
|
|
rng = np.random.default_rng(5)
|
|
OUTAGE = {"2024-07", "2025-02"} # 전산 장애로 기록이 올라오지 않은 월(가상)
|
|
rows, rid = [], 300000
|
|
for r in hist[hist["t"] >= 24].itertuples():
|
|
if r.ym in OUTAGE or r.demand == 0:
|
|
continue
|
|
k = int(np.ceil(r.demand / 5))
|
|
for j, q in enumerate(rng.multinomial(r.demand, np.ones(k) / k)):
|
|
if q == 0:
|
|
continue
|
|
rid += 1
|
|
rows.append({"req_id": f"R{rid}", "date": f"{r.ym}-{rng.integers(1, 29):02d}", "nsn": nsn_of[r.item_id],
|
|
"qty_req": int(q), "qty_issued": int(q), "uoi": "EA", "emergency": bool(j < r.emergency)})
|
|
raw = pd.DataFrame(rows)
|
|
n = len(raw)
|
|
idx = lambda frac: rng.choice(n, int(n * frac), replace=False)
|
|
raw.loc[idx(0.05), "nsn"] = raw["nsn"].str.replace("-", "") # 하이픈 누락
|
|
raw.loc[idx(0.03), "nsn"] = " " + raw["nsn"] + " " # 앞뒤 공백
|
|
raw.loc[idx(0.04), "date"] = raw["date"].str.replace("-", ".") # 날짜 형식 혼재
|
|
raw.loc[idx(0.02), "qty_req"] = -raw["qty_req"] # 음수(입력 오류)
|
|
raw.loc[idx(0.02), "qty_issued"] = raw["qty_req"].abs() + 2 # 불출 > 청구
|
|
bx = idx(0.02); raw.loc[bx, "uoi"] = "BX"; raw.loc[bx, "qty_req"] = 1; raw.loc[bx, "qty_issued"] = 1 # 상자(10EA) 단위
|
|
raw.loc[idx(0.01), "qty_req"] = np.nan # 수량 결측
|
|
raw.loc[idx(0.005), "date"] = "2027-01-15" # 미래 일자
|
|
raw.loc[idx(0.005), "nsn"] = "9999-00-000-0000" # 미등록 NSN
|
|
typo = [i for i in idx(0.01) if not raw.at[i, "emergency"]]
|
|
raw.loc[typo, "qty_req"] = raw.loc[typo, "qty_req"] * 10 # 0 하나 더 입력한 오타
|
|
raw = pd.concat([raw, raw.sample(60, random_state=1)], ignore_index=True) # 중복 전송
|
|
print(f"원시 청구 기록 {len(raw)}건 (가상), 전산 장애 월 {sorted(OUTAGE)}")
|
|
|
|
|
|
# 2) 정제 규칙 10종
|
|
def fmt_nsn(s):
|
|
d = re.sub(r"\D", "", str(s))
|
|
return f"{d[:4]}-{d[4:6]}-{d[6:9]}-{d[9:]}" if len(d) == 13 else str(s).strip()
|
|
|
|
|
|
log = {}
|
|
df = raw.copy()
|
|
before = len(df); df = df.drop_duplicates("req_id"); log["R1 중복 청구번호 제거"] = before - len(df)
|
|
em0 = int(df["emergency"].sum())
|
|
fixed = df["nsn"].map(fmt_nsn); log["R2 NSN 형식 정규화(공백·하이픈)"] = int((fixed != df["nsn"]).sum()); df["nsn"] = fixed
|
|
unk = ~df["nsn"].isin(master); log["R3 미등록 NSN 격리"] = int(unk.sum()); em_drop = int(df.loc[unk, "emergency"].sum()); df = df[~unk]
|
|
d2 = df["date"].str.replace(".", "-", regex=False); log["R4 날짜 형식 통일"] = int((d2 != df["date"]).sum()); df["date"] = pd.to_datetime(d2)
|
|
fut = df["date"] > "2025-12-31"; log["R5 미래 일자 제거"] = int(fut.sum()); em_drop += int(df.loc[fut, "emergency"].sum()); df = df[~fut]
|
|
neg = df["qty_req"] < 0; log["R6 음수 수량 부호 보정"] = int(neg.sum()); df.loc[neg, "qty_req"] = -df.loc[neg, "qty_req"]
|
|
na = df["qty_req"].isna(); log["R7 청구 수량 결측 -> 불출 수량으로 보완"] = int(na.sum()); df.loc[na, "qty_req"] = df.loc[na, "qty_issued"]
|
|
bxm = df["uoi"] == "BX"; log["R8 단위 환산(BX=10EA)"] = int(bxm.sum())
|
|
df.loc[bxm, ["qty_req", "qty_issued"]] = df.loc[bxm, ["qty_req", "qty_issued"]] * 10; df.loc[bxm, "uoi"] = "EA"
|
|
over = df["qty_issued"] > df["qty_req"]; log["R9 불출>청구 -> 청구 수량으로 제한"] = int(over.sum())
|
|
df.loc[over, "qty_issued"] = df.loc[over, "qty_req"]
|
|
cap = df.groupby("nsn")["qty_req"].transform(lambda s: s.median() * 5)
|
|
out = (df["qty_req"] > cap) & ~df["emergency"]
|
|
log["R10 오타성 이상치 보정(긴급 불출은 보존)"] = int(out.sum()); df.loc[out, "qty_req"] = (df.loc[out, "qty_req"] / 10).round()
|
|
df["qty_req"] = df["qty_req"].astype(int)
|
|
|
|
# 3) 월별 집계 + 결측 월 계절성 보간(R11): 작년 같은 달 x (최근 수준 / 작년 수준)
|
|
df["ym"] = df["date"].dt.strftime("%Y-%m")
|
|
monthly = df.pivot_table(index="ym", columns="nsn", values="qty_req", aggfunc="sum")
|
|
all_m = [f"{y}-{m:02d}" for y in (2024, 2025) for m in range(1, 13)]
|
|
monthly = monthly.reindex(all_m)
|
|
prev = hist[hist["t"].between(12, 23)].assign(nsn=lambda x: x["item_id"].map(nsn_of)).pivot(index="month", columns="nsn", values="demand")
|
|
seasonal_fill = monthly.copy()
|
|
for ym in OUTAGE:
|
|
i = all_m.index(ym)
|
|
ratio = monthly.iloc[max(0, i - 3):i].mean() / prev.loc[[(i - k) % 12 + 1 for k in (1, 2, 3)]].mean().replace(0, np.nan)
|
|
seasonal_fill.loc[ym] = (prev.loc[int(ym[-2:])] * ratio.fillna(1)).round()
|
|
linear_fill = monthly.interpolate()
|
|
monthly = monthly.fillna(0)
|
|
seasonal_fill = seasonal_fill.fillna(0)
|
|
log["R11 결측 월 계절성 보간"] = len(OUTAGE) * monthly.shape[1]
|
|
truth = hist[hist["t"] >= 24].assign(nsn=lambda x: x["item_id"].map(nsn_of)).pivot(index="ym", columns="nsn", values="demand")
|
|
err_s = np.abs(seasonal_fill.loc[sorted(OUTAGE)] - truth.loc[sorted(OUTAGE)]).values.mean()
|
|
err_l = np.abs(linear_fill.loc[sorted(OUTAGE)] - truth.loc[sorted(OUTAGE)]).values.mean()
|
|
|
|
print("\n[정제 규칙 적용 결과]")
|
|
for k, v in log.items():
|
|
print(f" {k:32s} {v:5d}건")
|
|
print(f"결측 월 보간 오차(MAE): 계절성 보간 {err_s:.2f}개 / 단순 선형 보간 {err_l:.2f}개")
|
|
|
|
# 4) 정합성 검증
|
|
checks = {"청구번호 유일": df["req_id"].is_unique, "NSN 형식": df["nsn"].str.match(r"^\d{4}-\d{2}-\d{3}-\d{4}$").all(),
|
|
"마스터 등록 NSN": df["nsn"].isin(master).all(), "수량 양수": (df["qty_req"] > 0).all(),
|
|
"불출<=청구": (df["qty_issued"] <= df["qty_req"]).all(), "단위 EA 통일": (df["uoi"] == "EA").all(),
|
|
"미래 일자 없음": (df["date"] <= "2025-12-31").all(), "결측 월 없음": not seasonal_fill.isna().any().any(),
|
|
"긴급 불출 보존": int(df["emergency"].sum()) == em0 - em_drop,
|
|
"월 합계 = 기록 합계": int(monthly.values.sum()) == int(df["qty_req"].sum())}
|
|
rate = sum(checks.values()) / len(checks)
|
|
print(f"\n정합성 검증 {sum(checks.values())}/{len(checks)} 통과({rate:.0%}) - 목표 100%: {'달성' if rate == 1 else '미달'}, 적용 규칙 {len(log)}종(목표 10종 이상)")
|
|
|
|
# 5) 전후 기술통계 비교
|
|
stat = pd.DataFrame({"정제 전": pd.to_numeric(raw["qty_req"], errors="coerce").describe(), "정제 후": df["qty_req"].describe()}).round(2)
|
|
print("\n[청구 수량 기술통계 전후 비교]")
|
|
print(stat.T.to_string())
|
|
|
|
# 6) 계절성 분리(고전적 분해): 추세 = 12개월 중심 이동평균, 계절지수 = 월별 (실적/추세) 평균
|
|
bat = items[items["fsc"] == "6140"].sort_values("base")["nsn"].iloc[-1] # 수요가 가장 많은 축전지 품목
|
|
series = pd.concat([truth.iloc[:0], seasonal_fill])[bat].astype(float)
|
|
full = hist[hist["item_id"] == items.set_index("nsn").loc[bat, "item_id"]].set_index("ym")["demand"].astype(float)
|
|
full.loc[series.index] = series
|
|
trend = full.rolling(12, center=True).mean()
|
|
sidx = (full / trend).groupby([int(i[-2:]) for i in full.index]).mean()
|
|
sidx = sidx / sidx.mean()
|
|
print(f"\n축전지 품목 {bat} 계절지수(1=평균):", " ".join(f"{m}월 {v:.2f}" for m, v in sidx.items()))
|
|
print("해석: 겨울(12~2월) 축전지 수요 증가와 훈련 월(3·8·10월) 증가가 함께 나타난다. 훈련 효과는 미션 4에서 별도 피처로 분리한다.")
|
|
|
|
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
|
|
ax[0].plot(full.values, label="demand"); ax[0].plot(trend.values, label="trend (12M centered MA)")
|
|
ax[0].set_title(f"Battery item {bat}: demand vs trend"); ax[0].set_xlabel("month index"); ax[0].legend()
|
|
ax[1].bar(sidx.index, sidx.values); ax[1].axhline(1, c="k", ls="--"); ax[1].set_title("Seasonal index by month"); ax[1].set_xlabel("month")
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
원시 청구 기록 4918건 (가상), 전산 장애 월 ['2024-07', '2025-02']
|
|
|
|
[정제 규칙 적용 결과]
|
|
R1 중복 청구번호 제거 60건
|
|
R2 NSN 형식 정규화(공백·하이픈) 372건
|
|
R3 미등록 NSN 격리 24건
|
|
R4 날짜 형식 통일 191건
|
|
R5 미래 일자 제거 24건
|
|
R6 음수 수량 부호 보정 94건
|
|
R7 청구 수량 결측 -> 불출 수량으로 보완 47건
|
|
R8 단위 환산(BX=10EA) 97건
|
|
R9 불출>청구 -> 청구 수량으로 제한 96건
|
|
R10 오타성 이상치 보정(긴급 불출은 보존) 35건
|
|
R11 결측 월 계절성 보간 80건
|
|
결측 월 보간 오차(MAE): 계절성 보간 5.72개 / 단순 선형 보간 7.86개
|
|
|
|
정합성 검증 10/10 통과(100%) - 목표 100%: 달성, 적용 규칙 11종(목표 10종 이상)
|
|
|
|
[청구 수량 기술통계 전후 비교]
|
|
count mean std min 25% 50% 75% max
|
|
정제 전 4870.0 4.75 4.76 -12.0 3.0 4.0 6.0 80.0
|
|
정제 후 4810.0 4.80 2.17 1.0 3.0 5.0 6.0 20.0
|
|
|
|
축전지 품목 6140-00-643-1860 계절지수(1=평균): 1월 1.39 2월 1.16 3월 1.13 4월 0.69 5월 0.69 6월 0.66 7월 0.75 8월 1.16 9월 0.85 10월 1.38 11월 0.68 12월 1.46
|
|
해석: 겨울(12~2월) 축전지 수요 증가와 훈련 월(3·8·10월) 증가가 함께 나타난다. 훈련 효과는 미션 4에서 별도 피처로 분리한다.
|
|
```
|
|
|
|
## 미션 3. 이동평균 및 지수평활법 기반 1차 부품 소요량 예측 Baseline 알고리즘 구현
|
|
|
|
- 난이도: 중급 / 예상 소요시간: 2.5시간
|
|
- 코드 설명: 이동평균(3·6개월)과 단순 지수평활법으로 품목별 다음 달 부품 소요량을 예측하는 Baseline을 구현하고, 탐색적 분석(EDA) 차트 8종·상관 검정·ABC(파레토) 분류로 수요 특성과 핵심 관리 품목을 도출한다.
|
|
|
|
```python
|
|
"""이동평균(3·6개월)과 단순 지수평활법으로 품목별 다음 달 부품 소요량을 예측하는 Baseline을 구현하고, 탐색적 분석(EDA) 차트 8종·상관 검정·ABC(파레토) 분류로 수요 특성과 핵심 관리 품목을 도출한다."""
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy import stats
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
items = make_items()
|
|
hist = make_history(items).merge(items[["item_id", "group", "price", "status", "lead_time"]], on="item_id")
|
|
TEST_FROM = 42 # 마지막 6개월(가상 2025-07 ~ 2025-12)을 평가 구간으로 둔다
|
|
|
|
|
|
def ses(y, alpha):
|
|
"""단순 지수평활: 다음 달 예측 = alpha x 이번 달 실적 + (1 - alpha) x 이번 달 예측."""
|
|
f = np.empty(len(y)); f[0] = y[:6].mean()
|
|
for i in range(1, len(y)):
|
|
f[i] = alpha * y[i - 1] + (1 - alpha) * f[i - 1]
|
|
return f
|
|
|
|
|
|
# 1) Baseline 예측(한 달 앞): MA3, MA6, SES(품목별 alpha를 학습 구간에서 선택), 작년 같은 달
|
|
preds = []
|
|
for iid, g in hist.groupby("item_id"):
|
|
y = g.sort_values("t")["demand"].to_numpy(float)
|
|
best = min([0.1, 0.2, 0.3, 0.5], key=lambda a: np.abs(ses(y, a)[12:TEST_FROM] - y[12:TEST_FROM]).mean())
|
|
s = pd.Series(y)
|
|
preds.append(pd.DataFrame({"item_id": iid, "t": np.arange(len(y)), "y": y, "MA3": s.shift(1).rolling(3).mean(),
|
|
"MA6": s.shift(1).rolling(6).mean(), "SES": ses(y, best), "작년 같은 달": s.shift(12)}))
|
|
P = pd.concat(preds)
|
|
T = P[P["t"] >= TEST_FROM]
|
|
ab = items[items["base"] >= 10]["item_id"] # 수요가 충분한 품목(MAPE 해석 가능)
|
|
rows = []
|
|
for m in ["MA3", "MA6", "SES", "작년 같은 달"]:
|
|
e = T["y"] - T[m]
|
|
big = T[T["item_id"].isin(ab)]
|
|
rows.append({"방법": m, "MAE(개)": round(e.abs().mean(), 2), "WAPE": f"{e.abs().sum() / T['y'].sum():.1%}",
|
|
"MAPE(수요 많은 품목)": f"{(np.abs(big['y'] - big[m]) / big['y'].clip(lower=1)).mean():.1%}"})
|
|
base_tbl = pd.DataFrame(rows)
|
|
print("[Baseline 예측 성능 - 평가 구간 6개월]")
|
|
print(base_tbl.to_string(index=False))
|
|
print("해석: 이동평균은 훈련 월 급증과 계절 변화를 한 달 늦게 따라가 오차가 커진다(미션 4에서 개선).")
|
|
|
|
# 2) 상관 검정: 품목 평균 대비 수요 비율과 후보 피처
|
|
hist = hist.sort_values(["item_id", "t"])
|
|
hist["ratio"] = hist["demand"] / hist.groupby("item_id")["demand"].transform("mean")
|
|
g = hist.groupby("item_id")["ratio"]
|
|
hist["lag1_ratio"] = g.shift(1); hist["lag12_ratio"] = g.shift(12)
|
|
hist["winter"] = hist["month"].isin([12, 1, 2]).astype(int)
|
|
cand = ["op_hours", "exercise", "availability", "lag12_ratio", "lag1_ratio", "t", "winter", "price"]
|
|
res = []
|
|
for c in cand:
|
|
d = hist[["ratio", c]].dropna()
|
|
r, p = stats.pearsonr(d[c], d["ratio"])
|
|
res.append({"피처": c, "상관계수": round(r, 3), "p-value": f"{p:.1e}", "유의(p<0.05)": "예" if p < 0.05 else "아니오"})
|
|
res = pd.DataFrame(res)
|
|
n_sig = (res["유의(p<0.05)"] == "예").sum()
|
|
print("\n[수요 비율과 피처의 상관 검정]")
|
|
print(res.to_string(index=False))
|
|
print(f"유의한 피처 {n_sig}개 (목표 5개 이상: {'달성' if n_sig >= 5 else '미달'})")
|
|
t_stat, p_ex = stats.ttest_ind(hist[hist["exercise"] == 1]["ratio"], hist[hist["exercise"] == 0]["ratio"])
|
|
print(f"훈련 월과 평시 월 수요 비율 차이 t검정: t={t_stat:.1f}, p={p_ex:.1e}")
|
|
|
|
# 3) ABC(파레토) 분류: 최근 12개월 수요 금액 기준
|
|
val = hist[hist["t"] >= 36].groupby("item_id").apply(lambda x: (x["demand"] * x["price"]).sum(), include_groups=False).sort_values(ascending=False)
|
|
cum = val.cumsum() / val.sum()
|
|
abc = pd.Series(np.where(cum <= 0.8, "A", np.where(cum <= 0.95, "B", "C")), index=val.index)
|
|
top = val.index[: int(np.ceil(len(val) * 0.2))]
|
|
print(f"\nABC 분류: A {sum(abc == 'A')}개 / B {sum(abc == 'B')}개 / C {sum(abc == 'C')}개, 상위 20% {len(top)}개 품목이 수요 금액의 {val[top].sum() / val.sum():.0%} 차지")
|
|
st = items.set_index("item_id").loc[top]
|
|
print("상위 20% 중 단종예고·단종 품목:", ", ".join(f"{i}({s})" for i, s in st["status"].items() if s != "생산중") or "없음")
|
|
|
|
# 4) EDA 차트 8종
|
|
fig, ax = plt.subplots(2, 4, figsize=(18, 8))
|
|
tot = hist.groupby("t")["demand"].sum()
|
|
ax[0, 0].plot(tot.index, tot.values); ax[0, 0].set_title("1. Total monthly demand")
|
|
hist.groupby(["t", "group"])["demand"].sum().unstack().plot(ax=ax[0, 1], legend=False); ax[0, 1].set_title("2. Demand by FSC group")
|
|
hist.boxplot(column="ratio", by="month", ax=ax[0, 2], grid=False); ax[0, 2].set_title("3. Seasonality (demand ratio)")
|
|
ax[0, 3].scatter(hist["op_hours"], hist["ratio"], s=4, alpha=0.3); ax[0, 3].set_title("4. Operating hours vs demand ratio")
|
|
hist.groupby("exercise")["ratio"].mean().plot.bar(ax=ax[1, 0]); ax[1, 0].set_title("5. Exercise month effect")
|
|
ax[1, 1].plot(np.arange(1, len(cum) + 1), cum.values, "o-", ms=3); ax[1, 1].axhline(0.8, ls="--", c="gray"); ax[1, 1].set_title("6. Pareto curve (ABC)")
|
|
items.assign(st=items["status"].map({"생산중": "active", "단종예고": "EOL notice", "단종": "discontinued"})).boxplot(column="lead_time", by="st", ax=ax[1, 2], grid=False); ax[1, 2].set_title("7. Lead time by production status")
|
|
ag = hist.groupby("t").agg(av=("availability", "mean"), d=("demand", "sum"))
|
|
ax[1, 3].scatter(ag["av"], ag["d"]); ax[1, 3].set_title("8. Availability vs total demand")
|
|
for a in ax.ravel():
|
|
a.set_xlabel(""); a.tick_params(labelsize=7)
|
|
plt.suptitle("MNT02 EDA - 8 charts (synthetic data)"); plt.tight_layout(); plt.show()
|
|
print("EDA 차트 8종 생성 완료. 핵심 발견: 운용시간·훈련 월이 수요를 끌어올리고, 단종 품목은 리드타임이 길어 결품 위험이 크다.")
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[Baseline 예측 성능 - 평가 구간 6개월]
|
|
방법 MAE(개) WAPE MAPE(수요 많은 품목)
|
|
MA3 7.34 28.7% 32.7%
|
|
MA6 7.09 27.7% 31.6%
|
|
SES 6.53 25.5% 28.9%
|
|
작년 같은 달 5.38 21.0% 22.8%
|
|
해석: 이동평균은 훈련 월 급증과 계절 변화를 한 달 늦게 따라가 오차가 커진다(미션 4에서 개선).
|
|
|
|
[수요 비율과 피처의 상관 검정]
|
|
피처 상관계수 p-value 유의(p<0.05)
|
|
op_hours 0.477 8.3e-110 예
|
|
exercise 0.470 4.4e-106 예
|
|
availability -0.450 1.6e-96 예
|
|
lag12_ratio 0.339 3.7e-40 예
|
|
lag1_ratio -0.111 1.5e-06 예
|
|
t 0.066 3.8e-03 예
|
|
winter -0.032 1.6e-01 아니오
|
|
price 0.000 1.0e+00 아니오
|
|
유의한 피처 6개 (목표 5개 이상: 달성)
|
|
훈련 월과 평시 월 수요 비율 차이 t검정: t=23.3, p=4.4e-106
|
|
|
|
ABC 분류: A 13개 / B 13개 / C 14개, 상위 20% 8개 품목이 수요 금액의 67% 차지
|
|
상위 20% 중 단종예고·단종 품목: P016(단종), P028(단종), P032(단종예고), P007(단종예고)
|
|
EDA 차트 8종 생성 완료. 핵심 발견: 운용시간·훈련 월이 수요를 끌어올리고, 단종 품목은 리드타임이 길어 결품 위험이 크다.
|
|
```
|
|
|
|
## 미션 4. 시계열 머신러닝(Prophet/LSTM) 기반 결전 대비 야전 부품 수요 예측 모델 고도화
|
|
|
|
- 난이도: 중급 / 예상 소요시간: 3.0시간
|
|
- 코드 설명: 과거 수요 지연값·이동평균·계절지수·훈련 계획·운용시간 등 파생 피처 15개로 시계열 머신러닝 수요 예측 모델(Prophet/LSTM 대신 포아송 그래디언트 부스팅)을 만들고, 시계열 교차검증으로 MAPE를 확인한 뒤 작전 템포 증가(결전 대비) 시 수요를 예측한다.
|
|
|
|
```python
|
|
"""과거 수요 지연값·이동평균·계절지수·훈련 계획·운용시간 등 파생 피처 15개로 시계열 머신러닝 수요 예측 모델(Prophet/LSTM 대신 포아송 그래디언트 부스팅)을 만들고, 시계열 교차검증으로 MAPE를 확인한 뒤 작전 템포 증가(결전 대비) 시 수요를 예측한다."""
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from sklearn.ensemble import HistGradientBoostingRegressor
|
|
from sklearn.inspection import permutation_importance
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
df = make_features(hist)
|
|
ab = set(items[items["base"] >= 10]["item_id"]) # 수요가 충분한 품목(MAPE 평가 대상)
|
|
print(f"파생 피처 {len(FEATS)}개 (목표 10개 이상): {', '.join(FEATS)}")
|
|
|
|
# 1) 시계열 교차검증(확장 창): 과거로 학습 -> 바로 다음 6개월 평가, 3회 반복
|
|
rows = []
|
|
for cut in (30, 36, 42):
|
|
tr, te = df[df["t"] < cut], df[(df["t"] >= cut) & (df["t"] < cut + 6)]
|
|
p = new_model().fit(tr[FEATS], tr["demand"]).predict(te[FEATS])
|
|
big = te["item_id"].isin(ab).to_numpy()
|
|
rows.append({"학습 구간": f"t<{cut}", "평가 구간": f"t={cut}~{cut + 5}",
|
|
"ML MAE": round(np.abs(te["demand"] - p).mean(), 2), "MA3 MAE": round(np.abs(te["demand"] - te["ma3"]).mean(), 2),
|
|
"ML MAPE": mape(te["demand"][big], p[big]), "MA3 MAPE": mape(te["demand"][big], te["ma3"][big])})
|
|
cv = pd.DataFrame(rows)
|
|
print("\n[시계열 교차검증 결과 - MAPE는 수요 많은 품목 기준]")
|
|
print(cv.assign(**{"ML MAPE": cv["ML MAPE"].map("{:.1%}".format), "MA3 MAPE": cv["MA3 MAPE"].map("{:.1%}".format)}).to_string(index=False))
|
|
m_mape = cv["ML MAPE"].mean()
|
|
print(f"평균 MAPE: ML {m_mape:.1%} / MA3 {cv['MA3 MAPE'].mean():.1%} -> 목표(MAPE 25% 이하): {'달성' if m_mape <= 0.25 else '미달'}")
|
|
|
|
# 2) 최종 모델(t<42 학습)과 피처 중요도(순열 중요도)
|
|
tr, te = df[df["t"] < 42], df[df["t"] >= 42]
|
|
model = new_model().fit(tr[FEATS], tr["demand"])
|
|
imp = permutation_importance(model, te[FEATS], te["demand"], n_repeats=5, random_state=0, scoring="neg_mean_absolute_error")
|
|
imp = pd.Series(imp.importances_mean, index=FEATS).sort_values(ascending=False)
|
|
print("\n[피처 중요도 상위 6개(MAE 증가량)]", ", ".join(f"{k} {v:.2f}" for k, v in imp.head(6).items()))
|
|
print("중요도가 0에 가까운 피처(차원 축소 후보):", ", ".join(imp[imp < 0.02].index) or "없음")
|
|
|
|
# 3) 결전 대비 시나리오: 다음 달(가상 2026-01) 운용시간 1.5배 + 훈련 월 지정
|
|
future = []
|
|
for it in items.itertuples():
|
|
last = hist[(hist["item_id"] == it.item_id) & (hist["t"] == 47)].iloc[0]
|
|
future.append({**last.to_dict(), "t": 48, "ym": "2026-01", "month": 1, "demand": 0, "exercise": 0})
|
|
fut = pd.DataFrame(future)
|
|
fut["op_hours"] = hist[hist["t"] == 36].set_index("item_id").loc[fut["item_id"], "op_hours"].to_numpy() # 작년 1월 계획 운용시간
|
|
res = {}
|
|
for name, mult, ex in [("평시 계획", 1.0, 0), ("작전 템포 1.5배", 1.5, 1)]:
|
|
f2 = fut.assign(op_hours=fut["op_hours"] * mult, exercise=ex)
|
|
X = make_features(pd.concat([hist, f2], ignore_index=True))
|
|
X = X[X["t"] == 48]
|
|
res[name] = pd.Series(model.predict(X[FEATS]), index=X["item_id"])
|
|
comp = pd.DataFrame(res).round(1)
|
|
comp["증가율"] = (comp["작전 템포 1.5배"] / comp["평시 계획"] - 1).map("{:+.0%}".format)
|
|
comp = comp.join(items.set_index("item_id")[["group", "status"]])
|
|
print("\n[결전 대비 다음 달 수요 예측 - 증가량 상위 6개 품목]")
|
|
print(comp.loc[(comp["작전 템포 1.5배"] - comp["평시 계획"]).sort_values(ascending=False).index[:6]].to_string())
|
|
print(f"전체 예측 수요: 평시 {comp['평시 계획'].sum():.0f}개 -> 작전 템포 1.5배 {comp['작전 템포 1.5배'].sum():.0f}개")
|
|
|
|
# 4) 예측 그림: 수요 많은 품목 1개
|
|
iid = items.sort_values("base")["item_id"].iloc[-1]
|
|
s = te[te["item_id"] == iid]
|
|
full = df[df["item_id"] == iid]
|
|
plt.figure(figsize=(9, 4))
|
|
plt.plot(full["t"], full["demand"], "k.-", label="actual")
|
|
plt.plot(s["t"], model.predict(s[FEATS]), "r-o", label="ML forecast")
|
|
plt.plot(s["t"], s["ma3"], "b--", label="MA3 baseline")
|
|
plt.axvline(41.5, c="gray", ls=":"); plt.xlabel("month index"); plt.ylabel("demand (EA)")
|
|
plt.title(f"{iid} one-month-ahead forecast (synthetic)"); plt.legend(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
파생 피처 15개 (목표 10개 이상): lag1, lag2, lag3, lag12, ma3, ma6, std6, season_idx, month_sin, month_cos, exercise, op_hours, op_ratio, item_level, trend_t
|
|
|
|
[시계열 교차검증 결과 - MAPE는 수요 많은 품목 기준]
|
|
학습 구간 평가 구간 ML MAE MA3 MAE ML MAPE MA3 MAPE
|
|
t<30 t=30~35 5.85 7.26 25.8% 30.9%
|
|
t<36 t=36~41 4.84 7.26 20.5% 30.9%
|
|
t<42 t=42~47 4.57 7.34 20.8% 32.7%
|
|
평균 MAPE: ML 22.3% / MA3 31.5% -> 목표(MAPE 25% 이하): 달성
|
|
|
|
[피처 중요도 상위 6개(MAE 증가량)] lag12 7.83, item_level 7.20, op_hours 0.95, std6 0.27, ma3 0.19, op_ratio 0.13
|
|
중요도가 0에 가까운 피처(차원 축소 후보): ma6, trend_t, lag2
|
|
|
|
[결전 대비 다음 달 수요 예측 - 증가량 상위 6개 품목]
|
|
평시 계획 작전 템포 1.5배 증가율 group status
|
|
item_id
|
|
P016 138.9 168.2 +21% gasket 단종
|
|
P028 74.1 99.5 +34% power_train 단종
|
|
P033 73.3 94.1 +28% fuel 생산중
|
|
P035 78.5 94.1 +20% brake_steer 생산중
|
|
P030 43.4 56.8 +31% brake_steer 단종예고
|
|
P026 15.6 24.3 +56% cooling 생산중
|
|
전체 예측 수요: 평시 946개 -> 작전 템포 1.5배 1210개
|
|
```
|
|
|
|
## 미션 5. 적정 안전재고량 산출 및 결품 위험 지수(Stockout Risk) 엔진 개발
|
|
|
|
- 난이도: 중급~고급 / 예상 소요시간: 3.5시간
|
|
- 코드 설명: 포아송 그래디언트 부스팅·랜덤포레스트 앙상블로 수요 예측을 고도화해 이동평균 기준 모델 대비 MAE/RMSE 개선율과 R2를 검증하고, 예측 오차로 품목별 적정 안전재고·재주문점을 산출한 뒤 리드타임 결품 확률과 단종 상태·단일 공급업체·대체품 부재를 결합한 결품 위험 지수 엔진을 만든다.
|
|
|
|
```python
|
|
"""포아송 그래디언트 부스팅·랜덤포레스트 앙상블로 수요 예측을 고도화해 이동평균 기준 모델 대비 MAE/RMSE 개선율과 R2를 검증하고, 예측 오차로 품목별 적정 안전재고·재주문점을 산출한 뒤 리드타임 결품 확률과 단종 상태·단일 공급업체·대체품 부재를 결합한 결품 위험 지수 엔진을 만든다."""
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy.stats import norm
|
|
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
|
|
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
# ---- 앙상블 수요 예측·안전재고·결품(단종) 위험 지수 공통 함수 ----
|
|
STATUS_W = {"생산중": 0.0, "단종예고": 0.5, "단종": 1.0} # [교육용가상자료] 단종 상태 가중치
|
|
|
|
|
|
def fit_ens(tr):
|
|
"""포아송 그래디언트 부스팅(70%) + 랜덤포레스트(30%) 앙상블."""
|
|
h = new_model().fit(tr[FEATS], tr["demand"])
|
|
r = RandomForestRegressor(n_estimators=150, min_samples_leaf=5, random_state=0, n_jobs=1).fit(tr[FEATS], tr["demand"])
|
|
return h, r
|
|
|
|
|
|
def pred_ens(m, X, w=0.7):
|
|
return np.clip(w * m[0].predict(X[FEATS]) + (1 - w) * m[1].predict(X[FEATS]), 0, None)
|
|
|
|
|
|
def next_month(hist, items, t_new=48, op_mult=1.0, ex=0):
|
|
"""다음 달(t_new) 예측용 행: 작년 같은 달 운용시간 x 배수, 훈련 여부 지정."""
|
|
last = hist[hist["t"] == t_new - 1].copy()
|
|
prev = hist[hist["t"] == t_new - 12].set_index("item_id")["op_hours"]
|
|
m = (t_new % 12) + 1
|
|
f = last.assign(t=t_new, ym=f"{2022 + t_new // 12}-{m:02d}", month=m, demand=0, exercise=ex,
|
|
op_hours=last["item_id"].map(prev).to_numpy() * op_mult)
|
|
X = make_features(pd.concat([hist, f], ignore_index=True))
|
|
return X[X["t"] == t_new].set_index("item_id").loc[items["item_id"]].reset_index()
|
|
|
|
|
|
def stock_policy(items, mu, sigma, z=1.65, review=1):
|
|
"""안전재고 = z x 월 예측오차 표준편차 x sqrt(리드타임+검토주기), 재주문점 = 월 수요 x (리드타임+검토주기) + 안전재고."""
|
|
L = items["lead_time"].to_numpy() + review
|
|
ss = z * sigma * np.sqrt(L)
|
|
return pd.DataFrame({"item_id": items["item_id"], "mu": mu, "sigma": sigma, "L": L,
|
|
"ss": np.ceil(ss), "rop": np.ceil(mu * L + ss)})
|
|
|
|
|
|
def risk_index(items, pol, stock):
|
|
"""결품 위험 지수(0~100): 리드타임 동안 수요가 가용 재고를 넘을 확률(정규 근사)을 기본으로,
|
|
단종 상태·단일 공급업체·대체품 부재에 가중을 준다(가상 가중치)."""
|
|
st = stock.set_index("item_id").loc[items["item_id"]]
|
|
avail = (st["on_hand"] + st["on_order"]).to_numpy()
|
|
p = 1 - norm.cdf((avail - pol["mu"].to_numpy() * pol["L"].to_numpy()) / (pol["sigma"].to_numpy() * np.sqrt(pol["L"].to_numpy()) + 1e-6))
|
|
sw = items["status"].map(STATUS_W).to_numpy()
|
|
sup = (items["suppliers"].to_numpy() == 1).astype(float)
|
|
alt = (~items["alt_available"].to_numpy()).astype(float)
|
|
score = 100 * np.clip(0.55 * p + 0.25 * sw + 0.10 * sup + 0.10 * alt, 0, 1)
|
|
return pd.DataFrame({"item_id": items["item_id"], "p_short": p, "status_w": sw, "single_sup": sup, "no_alt": alt,
|
|
"avail": avail, "risk": score.round(1)})
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
df = make_features(hist)
|
|
stock = make_stock(items, hist)
|
|
|
|
# 1) Hold-out(t=42~47) 성능 비교: 기준 모델(MA3) / 단일 모델 / 앙상블. 앙상블 가중치는 t=36~41 검증 구간에서 고른다
|
|
inner = fit_ens(df[df["t"] < 36]); vv = df[(df["t"] >= 36) & (df["t"] < 42)]
|
|
best_w = min((0.5, 0.6, 0.7, 0.8, 0.9, 1.0), key=lambda w: mean_absolute_error(vv["demand"], pred_ens(inner, vv, w)))
|
|
tr, te = df[df["t"] < 42], df[df["t"] >= 42].copy()
|
|
ens = fit_ens(tr)
|
|
te["ens"], te["hgb"] = pred_ens(ens, te, best_w), np.clip(ens[0].predict(te[FEATS]), 0, None)
|
|
rows = {}
|
|
for nm, col in [("기준 모델(MA3)", "ma3"), ("단일 포아송 GBM", "hgb"), (f"앙상블(w={best_w})", "ens")]:
|
|
rows[nm] = {"MAE": mean_absolute_error(te["demand"], te[col]), "RMSE": mean_squared_error(te["demand"], te[col]) ** 0.5,
|
|
"R2": r2_score(te["demand"], te[col])}
|
|
res = pd.DataFrame(rows).T
|
|
print("[Hold-out 6개월(t=42~47), 품목 40개]")
|
|
print(res.to_string(float_format=lambda x: f"{x:.3f}"))
|
|
b, e = res.iloc[0], res.iloc[2]
|
|
im, ir = 1 - e["MAE"] / b["MAE"], 1 - e["RMSE"] / b["RMSE"]
|
|
print(f"오차율 개선 MAE {im:.1%}, RMSE {ir:.1%} (목표 20% 이상: {'달성' if min(im, ir) >= 0.2 else '미달'}), R2 {e['R2']:.3f} (목표 0.85 이상: {'달성' if e['R2'] >= 0.85 else '미달'})")
|
|
spike = te["exercise"] == 1
|
|
print(f"불규칙 수요(훈련 월) MAE: 기준 {mean_absolute_error(te.loc[spike, 'demand'], te.loc[spike, 'ma3']):.2f} -> 앙상블 {mean_absolute_error(te.loc[spike, 'demand'], te.loc[spike, 'ens']):.2f}")
|
|
|
|
# 2) 안전재고·재주문점: 다음 달 예측과 Hold-out 품목별 오차 표준편차
|
|
X = next_month(hist, items)
|
|
mu = pred_ens(fit_ens(df), X, best_w)
|
|
err = (te["demand"] - te["ens"]).groupby(te["item_id"]).std().reindex(items["item_id"]).to_numpy()
|
|
sigma = np.maximum(err, np.sqrt(np.maximum(mu, 1))) # 오차 표준편차가 포아송 변동보다 작으면 포아송 기준 사용
|
|
pol = stock_policy(items, mu, sigma)
|
|
risk = risk_index(items, pol, stock)
|
|
out = items[["item_id", "group", "status", "lead_time", "suppliers"]].merge(pol, on="item_id").merge(risk, on="item_id")
|
|
print(f"\n[다음 달(가상 2026-01) 안전재고 정책] 서비스 수준 95%(z=1.65), 안전재고 합 {out['ss'].sum():.0f}개")
|
|
top = out.sort_values("risk", ascending=False).head(8)
|
|
show = top[["item_id", "group", "status", "lead_time", "mu", "ss", "rop", "avail", "p_short", "risk"]].copy()
|
|
show.columns = ["품목", "품목군", "생산상태", "LT(월)", "월수요", "안전재고", "재주문점", "가용재고", "결품확률", "위험지수"]
|
|
print("[결품 위험 지수 상위 8개]")
|
|
print(show.to_string(index=False, float_format=lambda x: f"{x:.2f}"))
|
|
grade = pd.cut(out["risk"], [-1, 30, 60, 101], labels=["관찰", "주의", "경보"]).value_counts()
|
|
print(f"위험 등급: 경보 {grade['경보']}개, 주의 {grade['주의']}개, 관찰 {grade['관찰']}개 / 재주문점 미만 품목 {(out['avail'] < out['rop']).sum()}개")
|
|
print("단종·단종예고 품목은 리드타임이 길고 공급업체가 1곳이라 같은 결품 확률에서도 위험 지수가 높게 나온다.")
|
|
|
|
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
|
|
ax[0].scatter(te["demand"], te["ma3"], s=8, alpha=0.5, label="MA3 baseline"); ax[0].scatter(te["demand"], te["ens"], s=8, alpha=0.5, label="ensemble")
|
|
mx = te["demand"].max(); ax[0].plot([0, mx], [0, mx], "k--", lw=1); ax[0].set_xlabel("actual"); ax[0].set_ylabel("forecast"); ax[0].legend()
|
|
ax[0].set_title("Hold-out actual vs forecast")
|
|
col = out["status"].map({"생산중": "tab:green", "단종예고": "tab:orange", "단종": "tab:red"})
|
|
ax[1].scatter(out["p_short"], out["risk"], c=col); ax[1].set_xlabel("shortage probability over lead time"); ax[1].set_ylabel("risk index")
|
|
ax[1].set_title("Stockout risk (green=active, orange=EOL notice, red=discontinued)")
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[Hold-out 6개월(t=42~47), 품목 40개]
|
|
MAE RMSE R2
|
|
기준 모델(MA3) 7.339 11.130 0.839
|
|
단일 포아송 GBM 4.572 6.112 0.952
|
|
앙상블(w=0.5) 4.541 6.059 0.952
|
|
오차율 개선 MAE 38.1%, RMSE 45.6% (목표 20% 이상: 달성), R2 0.952 (목표 0.85 이상: 달성)
|
|
불규칙 수요(훈련 월) MAE: 기준 9.88 -> 앙상블 4.94
|
|
|
|
[다음 달(가상 2026-01) 안전재고 정책] 서비스 수준 95%(z=1.65), 안전재고 합 926개
|
|
[결품 위험 지수 상위 8개]
|
|
품목 품목군 생산상태 LT(월) 월수요 안전재고 재주문점 가용재고 결품확률 위험지수
|
|
P018 battery 단종 4 17.48 23.00 111.00 71 0.88 93.60
|
|
P016 gasket 단종 12 148.02 95.00 2019.00 714 1.00 90.00
|
|
P028 power_train 단종 11 68.99 48.00 876.00 515 1.00 90.00
|
|
P004 filter 단종예고 6 7.65 16.00 70.00 12 1.00 87.50
|
|
P025 filter 단종예고 12 13.94 23.00 204.00 89 1.00 87.50
|
|
P030 brake_steer 단종예고 12 45.84 43.00 639.00 162 1.00 87.50
|
|
P038 fuel 단종예고 6 11.87 25.00 108.00 40 1.00 87.40
|
|
P019 filter 단종예고 8 11.47 30.00 133.00 59 0.99 87.10
|
|
위험 등급: 경보 13개, 주의 2개, 관찰 25개 / 재주문점 미만 품목 20개
|
|
단종·단종예고 품목은 리드타임이 길고 공급업체가 1곳이라 같은 결품 확률에서도 위험 지수가 높게 나온다.
|
|
```
|
|
|
|
## 미션 6. 야전 부품 재고 현황 및 긴급 조달 필요 품목 가시화 대시보드 구현
|
|
|
|
- 난이도: 중급~고급 / 예상 소요시간: 3.0시간
|
|
- 코드 설명: 야전 부품 재고 현황(가용재고·재주문점·결품 위험 지수)과 긴급 조달 필요 품목을 보여 주는 대시보드를 만들고, 품목마다 위험 지수를 끌어올린 요인을 기준값 대체(occlusion) 방식으로 분해해 기여도 상위 3개 인자와 자연어 설명문('리드타임 증가 원인은 단종' 등)을 자동 생성한다.
|
|
|
|
```python
|
|
"""야전 부품 재고 현황(가용재고·재주문점·결품 위험 지수)과 긴급 조달 필요 품목을 보여 주는 대시보드를 만들고, 품목마다 위험 지수를 끌어올린 요인을 기준값 대체(occlusion) 방식으로 분해해 기여도 상위 3개 인자와 자연어 설명문('리드타임 증가 원인은 단종' 등)을 자동 생성한다."""
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy.stats import norm
|
|
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
# ---- 앙상블 수요 예측·안전재고·결품(단종) 위험 지수 공통 함수 ----
|
|
STATUS_W = {"생산중": 0.0, "단종예고": 0.5, "단종": 1.0} # [교육용가상자료] 단종 상태 가중치
|
|
|
|
|
|
def fit_ens(tr):
|
|
"""포아송 그래디언트 부스팅(70%) + 랜덤포레스트(30%) 앙상블."""
|
|
h = new_model().fit(tr[FEATS], tr["demand"])
|
|
r = RandomForestRegressor(n_estimators=150, min_samples_leaf=5, random_state=0, n_jobs=1).fit(tr[FEATS], tr["demand"])
|
|
return h, r
|
|
|
|
|
|
def pred_ens(m, X, w=0.7):
|
|
return np.clip(w * m[0].predict(X[FEATS]) + (1 - w) * m[1].predict(X[FEATS]), 0, None)
|
|
|
|
|
|
def next_month(hist, items, t_new=48, op_mult=1.0, ex=0):
|
|
"""다음 달(t_new) 예측용 행: 작년 같은 달 운용시간 x 배수, 훈련 여부 지정."""
|
|
last = hist[hist["t"] == t_new - 1].copy()
|
|
prev = hist[hist["t"] == t_new - 12].set_index("item_id")["op_hours"]
|
|
m = (t_new % 12) + 1
|
|
f = last.assign(t=t_new, ym=f"{2022 + t_new // 12}-{m:02d}", month=m, demand=0, exercise=ex,
|
|
op_hours=last["item_id"].map(prev).to_numpy() * op_mult)
|
|
X = make_features(pd.concat([hist, f], ignore_index=True))
|
|
return X[X["t"] == t_new].set_index("item_id").loc[items["item_id"]].reset_index()
|
|
|
|
|
|
def stock_policy(items, mu, sigma, z=1.65, review=1):
|
|
"""안전재고 = z x 월 예측오차 표준편차 x sqrt(리드타임+검토주기), 재주문점 = 월 수요 x (리드타임+검토주기) + 안전재고."""
|
|
L = items["lead_time"].to_numpy() + review
|
|
ss = z * sigma * np.sqrt(L)
|
|
return pd.DataFrame({"item_id": items["item_id"], "mu": mu, "sigma": sigma, "L": L,
|
|
"ss": np.ceil(ss), "rop": np.ceil(mu * L + ss)})
|
|
|
|
|
|
def risk_index(items, pol, stock):
|
|
"""결품 위험 지수(0~100): 리드타임 동안 수요가 가용 재고를 넘을 확률(정규 근사)을 기본으로,
|
|
단종 상태·단일 공급업체·대체품 부재에 가중을 준다(가상 가중치)."""
|
|
st = stock.set_index("item_id").loc[items["item_id"]]
|
|
avail = (st["on_hand"] + st["on_order"]).to_numpy()
|
|
p = 1 - norm.cdf((avail - pol["mu"].to_numpy() * pol["L"].to_numpy()) / (pol["sigma"].to_numpy() * np.sqrt(pol["L"].to_numpy()) + 1e-6))
|
|
sw = items["status"].map(STATUS_W).to_numpy()
|
|
sup = (items["suppliers"].to_numpy() == 1).astype(float)
|
|
alt = (~items["alt_available"].to_numpy()).astype(float)
|
|
score = 100 * np.clip(0.55 * p + 0.25 * sw + 0.10 * sup + 0.10 * alt, 0, 1)
|
|
return pd.DataFrame({"item_id": items["item_id"], "p_short": p, "status_w": sw, "single_sup": sup, "no_alt": alt,
|
|
"avail": avail, "risk": score.round(1)})
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
df = make_features(hist)
|
|
stock = make_stock(items, hist)
|
|
X = next_month(hist, items)
|
|
mu = pred_ens(fit_ens(df), X)
|
|
level = X["item_level"].to_numpy() # 최근 12개월 평균 수요(평시 수준)
|
|
sigma = np.maximum(0.25 * mu, np.sqrt(np.maximum(mu, 1)))
|
|
|
|
|
|
def score(it, m, s, st):
|
|
return risk_index(it, stock_policy(it, m, s), st)["risk"].to_numpy()
|
|
|
|
|
|
base = score(items, mu, sigma, stock)
|
|
pol = stock_policy(items, mu, sigma)
|
|
# 각 요인을 '문제 없는 기준값'으로 바꿨을 때 위험 지수가 얼마나 내려가는지 = 그 요인의 기여도
|
|
ref_lt = int(items.loc[items["status"] == "생산중", "lead_time"].median())
|
|
st_ok = stock.copy(); st_ok["on_hand"] = np.maximum(st_ok["on_hand"], (pol["rop"] - st_ok["on_order"]).clip(lower=0).astype(int).to_numpy())
|
|
FACTORS = {
|
|
"조달 리드타임 장기화": lambda: score(items.assign(lead_time=np.minimum(items["lead_time"], ref_lt)), mu, sigma, stock),
|
|
"수요 증가 예측": lambda: score(items, np.minimum(mu, level), sigma, stock),
|
|
"가용재고 부족": lambda: score(items, mu, sigma, st_ok),
|
|
"예측 불확실성": lambda: score(items, mu, np.sqrt(np.maximum(mu, 1)), stock),
|
|
"단종(예고) 상태": lambda: score(items.assign(status="생산중"), mu, sigma, stock),
|
|
"단일 공급업체": lambda: score(items.assign(suppliers=np.maximum(items["suppliers"], 2)), mu, sigma, stock),
|
|
"대체품 부재": lambda: score(items.assign(alt_available=True), mu, sigma, stock),
|
|
}
|
|
C = pd.DataFrame({k: base - f() for k, f in FACTORS.items()}, index=items["item_id"]).clip(lower=0)
|
|
top3 = C.apply(lambda r: list(r.sort_values(ascending=False).index[:3]), axis=1)
|
|
cover = np.mean([len(t) == 3 for t in top3])
|
|
print(f"[품목 {len(items)}개] 위험 지수 기여도 상위 3개 인자 자동 산출률 {cover:.0%} (목표 100%: {'달성' if cover == 1 else '미달'})")
|
|
|
|
dash = items[["item_id", "group", "status", "lead_time"]].assign(risk=base, avail=(stock["on_hand"] + stock["on_order"]).to_numpy(),
|
|
rop=pol["rop"].to_numpy(), mu=mu.round(1))
|
|
dash["urgent"] = (dash["risk"] >= 60) & (dash["avail"] < dash["rop"])
|
|
CAUSE = {"조달 리드타임 장기화": "조달 리드타임이 길다", "단종(예고) 상태": "생산 중단(예고)으로 추가 조달이 어렵다",
|
|
"가용재고 부족": "보유·미입고 재고가 재주문점보다 적다", "수요 증가 예측": "다음 달 수요가 평시보다 늘 것으로 예측된다",
|
|
"예측 불확실성": "수요 변동이 커 안전재고가 많이 필요하다", "단일 공급업체": "공급업체가 1곳뿐이다", "대체품 부재": "대체품이 없다"}
|
|
|
|
|
|
def explain(iid):
|
|
r = dash.set_index("item_id").loc[iid]
|
|
f = top3[iid]
|
|
parts = ", ".join(f"{k}(+{C.loc[iid, k]:.0f})" for k in f)
|
|
return f"{iid}({r['group']}, {r['status']}) 위험 {r['risk']:.0f}: {parts}. 주된 원인은 {CAUSE[f[0]]}."
|
|
|
|
|
|
urg = dash[dash["urgent"]].sort_values("risk", ascending=False)
|
|
print(f"\n[긴급 조달 필요 품목 {len(urg)}개(위험 60 이상 & 가용재고<재주문점)] 자연어 설명")
|
|
for iid in urg["item_id"].head(7):
|
|
print(" -", explain(iid))
|
|
print("\n[요인별 평균 기여도]", ", ".join(f"{k} {v:.1f}" for k, v in C.mean().sort_values(ascending=False).items()))
|
|
g = dash.groupby("status")["risk"].agg(["count", "mean"]).round(1)
|
|
print("[생산 상태별 평균 위험 지수]", ", ".join(f"{s} {r['mean']}({int(r['count'])}개)" for s, r in g.iterrows()))
|
|
print("설명은 조달·대체품 검토의 근거 자료이며 발주·대체 결정은 정비장교가 확인 후 내린다.")
|
|
|
|
fig, ax = plt.subplots(1, 3, figsize=(17, 4.8))
|
|
d15 = dash.sort_values("risk", ascending=False).head(15)
|
|
ax[0].barh(d15["item_id"], d15["avail"], color=np.where(d15["urgent"], "tab:red", "tab:gray"), label="available")
|
|
ax[0].scatter(d15["rop"], d15["item_id"], c="k", marker="|", s=200, label="reorder point"); ax[0].invert_yaxis()
|
|
ax[0].set_xscale("symlog"); ax[0].set_title("Top-15 risk items: stock vs reorder point (red=urgent)"); ax[0].legend()
|
|
hm = dash.pivot_table(index="group", columns="status", values="risk", aggfunc="mean")
|
|
im = ax[1].imshow(hm.values, cmap="Reds", vmin=0, vmax=100, aspect="auto"); plt.colorbar(im, ax=ax[1])
|
|
ax[1].set_xticks(range(hm.shape[1])); ax[1].set_xticklabels(["active" if c == "생산중" else "EOL notice" if c == "단종예고" else "discontinued" for c in hm.columns])
|
|
ax[1].set_yticks(range(len(hm))); ax[1].set_yticklabels(hm.index); ax[1].set_title("Mean risk index by group/status")
|
|
i0 = urg["item_id"].iloc[0]; w = C.loc[i0].sort_values()
|
|
ax[2].barh(range(len(w)), w.values, color="tab:red"); ax[2].set_yticks(range(len(w)))
|
|
EN = {"조달 리드타임 장기화": "lead time", "수요 증가 예측": "demand up", "가용재고 부족": "low stock", "예측 불확실성": "uncertainty",
|
|
"단종(예고) 상태": "EOL status", "단일 공급업체": "single supplier", "대체품 부재": "no alternate"}
|
|
ax[2].set_yticklabels([EN[k] for k in w.index])
|
|
ax[2].set_title(f"Risk contribution: {i0}")
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[품목 40개] 위험 지수 기여도 상위 3개 인자 자동 산출률 100% (목표 100%: 달성)
|
|
|
|
[긴급 조달 필요 품목 13개(위험 60 이상 & 가용재고<재주문점)] 자연어 설명
|
|
- P018(battery, 단종) 위험 96: 가용재고 부족(+49), 단종(예고) 상태(+25), 단일 공급업체(+10). 주된 원인은 보유·미입고 재고가 재주문점보다 적다.
|
|
- P028(power_train, 단종) 위험 90: 조달 리드타임 장기화(+55), 가용재고 부족(+52), 단종(예고) 상태(+25). 주된 원인은 조달 리드타임이 길다.
|
|
- P016(gasket, 단종) 위험 90: 가용재고 부족(+52), 단종(예고) 상태(+25), 단일 공급업체(+10). 주된 원인은 보유·미입고 재고가 재주문점보다 적다.
|
|
- P004(filter, 단종예고) 위험 88: 가용재고 부족(+53), 단종(예고) 상태(+12), 단일 공급업체(+10). 주된 원인은 보유·미입고 재고가 재주문점보다 적다.
|
|
- P025(filter, 단종예고) 위험 88: 가용재고 부족(+53), 조달 리드타임 장기화(+39), 단종(예고) 상태(+12). 주된 원인은 보유·미입고 재고가 재주문점보다 적다.
|
|
- P038(fuel, 단종예고) 위험 88: 가용재고 부족(+52), 단종(예고) 상태(+12), 단일 공급업체(+10). 주된 원인은 보유·미입고 재고가 재주문점보다 적다.
|
|
- P030(brake_steer, 단종예고) 위험 88: 가용재고 부족(+52), 단종(예고) 상태(+12), 단일 공급업체(+10). 주된 원인은 보유·미입고 재고가 재주문점보다 적다.
|
|
|
|
[요인별 평균 기여도] 가용재고 부족 19.5, 단일 공급업체 4.8, 조달 리드타임 장기화 4.5, 단종(예고) 상태 4.4, 대체품 부재 3.8, 수요 증가 예측 0.3, 예측 불확실성 0.3
|
|
[생산 상태별 평균 위험 지수] 단종 92.0(3개), 단종예고 84.5(8개), 생산중 13.7(29개)
|
|
설명은 조달·대체품 검토의 근거 자료이며 발주·대체 결정은 정비장교가 확인 후 내린다.
|
|
```
|
|
|
|
## 미션 7. 부품 발주량 최종 확정을 위한 정비장교 HITL 승인/조정 워크플로우 설계
|
|
|
|
- 난이도: 고급 / 예상 소요시간: 3.5시간
|
|
- 코드 설명: 예측 기반 발주 정책(안전재고 포함 목표재고)과 단종 품목 대체품 전환을 12개월 재고 백테스트로 현행(최근 평균+고정 여유) 정책과 비교해 비용·재고 회전율·충족률을 산출하고, 다음 달 발주량을 예산 제약 선형계획(LP)으로 추천한 뒤 정비장교가 승인·조정·반려·보류로 최종 확정하는 HITL 워크플로우를 해시 체인 감사 로그와 함께 구현한다.
|
|
|
|
```python
|
|
"""예측 기반 발주 정책(안전재고 포함 목표재고)과 단종 품목 대체품 전환을 12개월 재고 백테스트로 현행(최근 평균+고정 여유) 정책과 비교해 비용·재고 회전율·충족률을 산출하고, 다음 달 발주량을 예산 제약 선형계획(LP)으로 추천한 뒤 정비장교가 승인·조정·반려·보류로 최종 확정하는 HITL 워크플로우를 해시 체인 감사 로그와 함께 구현한다."""
|
|
import hashlib
|
|
import json
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy.stats import norm
|
|
from scipy.optimize import linprog
|
|
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
# ---- 앙상블 수요 예측·안전재고·결품(단종) 위험 지수 공통 함수 ----
|
|
STATUS_W = {"생산중": 0.0, "단종예고": 0.5, "단종": 1.0} # [교육용가상자료] 단종 상태 가중치
|
|
|
|
|
|
def fit_ens(tr):
|
|
"""포아송 그래디언트 부스팅(70%) + 랜덤포레스트(30%) 앙상블."""
|
|
h = new_model().fit(tr[FEATS], tr["demand"])
|
|
r = RandomForestRegressor(n_estimators=150, min_samples_leaf=5, random_state=0, n_jobs=1).fit(tr[FEATS], tr["demand"])
|
|
return h, r
|
|
|
|
|
|
def pred_ens(m, X, w=0.7):
|
|
return np.clip(w * m[0].predict(X[FEATS]) + (1 - w) * m[1].predict(X[FEATS]), 0, None)
|
|
|
|
|
|
def next_month(hist, items, t_new=48, op_mult=1.0, ex=0):
|
|
"""다음 달(t_new) 예측용 행: 작년 같은 달 운용시간 x 배수, 훈련 여부 지정."""
|
|
last = hist[hist["t"] == t_new - 1].copy()
|
|
prev = hist[hist["t"] == t_new - 12].set_index("item_id")["op_hours"]
|
|
m = (t_new % 12) + 1
|
|
f = last.assign(t=t_new, ym=f"{2022 + t_new // 12}-{m:02d}", month=m, demand=0, exercise=ex,
|
|
op_hours=last["item_id"].map(prev).to_numpy() * op_mult)
|
|
X = make_features(pd.concat([hist, f], ignore_index=True))
|
|
return X[X["t"] == t_new].set_index("item_id").loc[items["item_id"]].reset_index()
|
|
|
|
|
|
def stock_policy(items, mu, sigma, z=1.65, review=1):
|
|
"""안전재고 = z x 월 예측오차 표준편차 x sqrt(리드타임+검토주기), 재주문점 = 월 수요 x (리드타임+검토주기) + 안전재고."""
|
|
L = items["lead_time"].to_numpy() + review
|
|
ss = z * sigma * np.sqrt(L)
|
|
return pd.DataFrame({"item_id": items["item_id"], "mu": mu, "sigma": sigma, "L": L,
|
|
"ss": np.ceil(ss), "rop": np.ceil(mu * L + ss)})
|
|
|
|
|
|
def risk_index(items, pol, stock):
|
|
"""결품 위험 지수(0~100): 리드타임 동안 수요가 가용 재고를 넘을 확률(정규 근사)을 기본으로,
|
|
단종 상태·단일 공급업체·대체품 부재에 가중을 준다(가상 가중치)."""
|
|
st = stock.set_index("item_id").loc[items["item_id"]]
|
|
avail = (st["on_hand"] + st["on_order"]).to_numpy()
|
|
p = 1 - norm.cdf((avail - pol["mu"].to_numpy() * pol["L"].to_numpy()) / (pol["sigma"].to_numpy() * np.sqrt(pol["L"].to_numpy()) + 1e-6))
|
|
sw = items["status"].map(STATUS_W).to_numpy()
|
|
sup = (items["suppliers"].to_numpy() == 1).astype(float)
|
|
alt = (~items["alt_available"].to_numpy()).astype(float)
|
|
score = 100 * np.clip(0.55 * p + 0.25 * sw + 0.10 * sup + 0.10 * alt, 0, 1)
|
|
return pd.DataFrame({"item_id": items["item_id"], "p_short": p, "status_w": sw, "single_sup": sup, "no_alt": alt,
|
|
"avail": avail, "risk": score.round(1)})
|
|
|
|
# ---- 월 단위 재고 백테스트(가상 비용 지수) ----
|
|
HOLD, ORDER_COST, SHORT_MULT = 0.02, 50_000, 0.6 # [교육용가상자료] 월 보관비율(단가 대비), 발주 1회 비용(원), 결품 1개 손실(단가 배수)
|
|
|
|
|
|
ALT_PREMIUM, ALT_LT = 1.2, 3 # [교육용가상자료] 대체품 전환 시 단가 배수, 대체품 조달 리드타임(월)
|
|
|
|
|
|
def simulate(items, demand, fc, policy, z=1.65, months=None, sigma=None, eol=True):
|
|
"""품목별 월 재고 모의. demand/fc: (품목 x 월) 실제·예측 수요. policy: 'current'(최근 12개월 평균 + 3개월 여유 고정) 또는 'ai'(예측 + 안전재고).
|
|
단종 품목은 원 품목 발주 불가. 'ai' 정책에서 eol=True이면 단종 품목 중 대체품이 있는 품목은 대체품(단가 x1.2, 리드타임 3개월)으로 전환 발주한다. 반환: 품목별 보관비·발주비·결품손실·불출량·평균재고."""
|
|
n, T = demand.shape
|
|
L = items["lead_time"].to_numpy(); price = items["price"].to_numpy().astype(float)
|
|
can_order = (items["status"] != "단종").to_numpy()
|
|
if policy == "ai" and eol:
|
|
sw = (~can_order) & items["alt_available"].to_numpy()
|
|
can_order = can_order | sw
|
|
L = np.where(sw, ALT_LT, L); price = np.where(sw, price * ALT_PREMIUM, price)
|
|
avg = demand[:, :1].repeat(T, 1) if months is None else months
|
|
on = np.round(avg[:, 0] * 2.0)
|
|
pipe = np.zeros((n, T + int(L.max()) + 1)) # 도착 예정 물량(품목 x 월)
|
|
for i in range(n):
|
|
pipe[i, :L[i]] = np.round(avg[i, 0] * (1.0 if can_order[i] else 0.3))
|
|
res = np.zeros((n, 6)); rows = np.arange(n)
|
|
for t in range(T):
|
|
on += pipe[:, t]
|
|
issue = np.minimum(on, demand[:, t]); short = demand[:, t] - issue; on -= issue
|
|
ip = on + pipe[:, t + 1:].sum(1)
|
|
if policy == "current":
|
|
target = avg[:, t] * (L + 3)
|
|
else:
|
|
s = sigma if sigma is not None else np.sqrt(np.maximum(fc[:, t], 1))
|
|
target = fc[:, t] * (L + 1) + z * s * np.sqrt(L + 1)
|
|
q = np.where(can_order & (ip < target), np.ceil(target - ip), 0)
|
|
pipe[rows, t + L] += q
|
|
res += np.c_[HOLD * price * on, ORDER_COST * (q > 0), SHORT_MULT * price * short, issue, on, short]
|
|
out = pd.DataFrame(res, columns=["hold", "order", "short", "issued", "inv", "short_qty"], index=items["item_id"])
|
|
out["inv"] /= T
|
|
return out
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
df = make_features(hist)
|
|
|
|
# 1) 백테스트(t=36~47, 12개월): 한 달 앞 예측은 t<36으로 학습한 앙상블 사용
|
|
m = fit_ens(df[df["t"] < 36])
|
|
bt = df[df["t"] >= 36].copy(); bt["fc"] = pred_ens(m, bt)
|
|
P = lambda c: bt.pivot(index="item_id", columns="t", values=c).loc[items["item_id"]].to_numpy()
|
|
D, F, A = P("demand"), P("fc"), P("item_level")
|
|
sig = np.maximum(0.25 * F.mean(1), np.sqrt(np.maximum(F.mean(1), 1)))
|
|
|
|
|
|
def summary(r, dem):
|
|
cost = r[["hold", "order", "short"]].sum()
|
|
return {"보관비": cost["hold"] / 1e6, "발주비": cost["order"] / 1e6, "결품손실": cost["short"] / 1e6, "총비용(백만)": cost.sum() / 1e6,
|
|
"충족률": 1 - r["short_qty"].sum() / dem.sum(), "회전율": r["issued"].sum() / r["inv"].sum()}
|
|
|
|
|
|
runs = {"현행(평균+3개월 여유)": simulate(items, D, F, "current", months=A),
|
|
"예측 정책(z=1.65)": simulate(items, D, F, "ai", 1.65, A, sig, eol=False),
|
|
"예측+단종 대체품 전환": simulate(items, D, F, "ai", 1.65, A, sig, eol=True)}
|
|
tab = pd.DataFrame({k: summary(v, D) for k, v in runs.items()}).T
|
|
print("[12개월 재고 백테스트(가상 비용 지수, 백만 원)]")
|
|
print(tab.to_string(formatters={"충족률": "{:.1%}".format, "회전율": "{:.2f}".format}, float_format=lambda x: f"{x:.1f}"))
|
|
c0, c2 = tab.iloc[0]["총비용(백만)"], tab.iloc[2]["총비용(백만)"]
|
|
t0, t1 = tab.iloc[0]["회전율"], tab.iloc[1]["회전율"]
|
|
print(f"총비용 절감 {1 - c2 / c0:.1%}, 예측 정책만의 회전율 향상 {t1 / t0 - 1:.1%} (목표: 비용 15% 절감 또는 회전율 20% 향상 -> {'달성' if (1 - c2 / c0) >= 0.15 or t1 / t0 - 1 >= 0.2 else '미달'})")
|
|
act = items["status"] != "단종"
|
|
a0, a1 = runs["현행(평균+3개월 여유)"][act.to_numpy()], runs["예측 정책(z=1.65)"][act.to_numpy()]
|
|
print(f"효과 분해: 생산중·단종예고 품목 비용 {1 - a1[['hold', 'order', 'short']].sum().sum() / a0[['hold', 'order', 'short']].sum().sum():.1%} 절감(예측 정책), 나머지는 단종 품목 대체품 전환 효과")
|
|
|
|
# 민감도: 서비스 수준(z)과 돌발 수요(최근 6개월 수요 1.5배, 가상 전시 전환)
|
|
D2 = D.copy(); D2[:, 6:] = np.round(D2[:, 6:] * 1.5)
|
|
sens = []
|
|
for z in (1.28, 1.65, 2.05):
|
|
for nm, dd in (("평시", D), ("돌발 1.5배", D2)):
|
|
r = summary(simulate(items, dd, F, "ai", z, A, sig), dd)
|
|
sens.append({"z": z, "수요": nm, "총비용": r["총비용(백만)"], "충족률": r["충족률"]})
|
|
print("[민감도] " + " / ".join(f"z={s['z']} {s['수요']}: {s['총비용']:.0f}백만, 충족 {s['충족률']:.1%}" for s in sens))
|
|
|
|
|
|
# 2) 다음 달 발주 추천: 예산 제약 LP (위험 지수 가중 부족분 충족 최대화)
|
|
class Audit:
|
|
def __init__(self):
|
|
self.log = []
|
|
|
|
def add(self, **body):
|
|
prev = self.log[-1]["hash"] if self.log else "GENESIS"
|
|
body = {"seq": len(self.log), **body}
|
|
self.log.append({**body, "prev": prev, "hash": hashlib.sha256((prev + json.dumps(body, sort_keys=True, ensure_ascii=False)).encode()).hexdigest()})
|
|
|
|
def verify(self):
|
|
prev = "GENESIS"
|
|
for e in self.log:
|
|
b = {k: v for k, v in e.items() if k not in ("prev", "hash")}
|
|
if e["prev"] != prev or hashlib.sha256((prev + json.dumps(b, sort_keys=True, ensure_ascii=False)).encode()).hexdigest() != e["hash"]:
|
|
return e["seq"]
|
|
prev = e["hash"]
|
|
return None
|
|
|
|
|
|
stock = make_stock(items, hist)
|
|
X = next_month(hist, items)
|
|
mu = pred_ens(fit_ens(df), X)
|
|
sg = np.maximum(0.25 * mu, np.sqrt(np.maximum(mu, 1)))
|
|
alt_sw = (items["status"] == "단종") & items["alt_available"]
|
|
it2 = items.assign(lead_time=np.where(alt_sw, ALT_LT, items["lead_time"]), price=np.where(alt_sw, items["price"] * ALT_PREMIUM, items["price"]))
|
|
pol = stock_policy(it2, mu, sg)
|
|
risk = risk_index(items, pol, stock)["risk"].to_numpy()
|
|
ip = (stock["on_hand"] + stock["on_order"]).to_numpy()
|
|
orderable = (items["status"] != "단종").to_numpy() | alt_sw.to_numpy()
|
|
gap = np.where(orderable, np.maximum(0, np.ceil(pol["rop"].to_numpy() - ip)), 0)
|
|
price = it2["price"].to_numpy().astype(float)
|
|
BUDGET = 0.6 * float((gap * price).sum())
|
|
idx = np.where(gap > 0)[0]
|
|
lp = linprog(c=-(risk[idx] + 10) / gap[idx], A_ub=[price[idx]], b_ub=[BUDGET], bounds=[(0, g) for g in gap[idx]], method="highs")
|
|
q = np.zeros(len(items)); q[idx] = np.floor(lp.x)
|
|
print(f"\n[다음 달 발주 추천 LP] 부족 품목 {len(idx)}개, 예산 {BUDGET / 1e6:.1f}백만(부족분 전액의 60%, 가상) 사용 {np.dot(q, price) / 1e6:.1f}백만, 추천 {int((q > 0).sum())}건")
|
|
|
|
audit, rows = Audit(), []
|
|
for k, i in enumerate(np.where(q > 0)[0]):
|
|
iid = items.loc[i, "item_id"]; kind = "대체품 전환" if alt_sw[i] else "정규 발주"
|
|
audit.add(t="추천", item=iid, kind=kind, qty=int(q[i]), risk=float(risk[i]), status="결재대기", by="AI")
|
|
# 가상 정비장교 판단(시뮬레이션): 근거 확인 후 승인·조정·반려, 일부는 시한 내 무응답(보류=발주하지 않음)
|
|
if kind == "대체품 전환":
|
|
dec, final, why = "반려", 0, "대체품 호환성 시험 결과 확인 후 재상신"
|
|
elif q[i] * price[i] > 0.15 * BUDGET:
|
|
dec, final, why = "조정", int(q[i] * 0.7), "고액 발주 분할(1차 70%), 다음 달 재검토"
|
|
elif k % 6 == 5:
|
|
dec, final, why = "보류", 0, "결재 시한 초과 -> 발주하지 않음"
|
|
else:
|
|
dec, final, why = "승인", int(q[i]), "위험 지수·재고 근거 확인"
|
|
audit.add(t="결재", item=iid, decision=dec, final_qty=final, reason=why, by="정비장교(가상)")
|
|
rows.append({"품목": iid, "구분": kind, "위험": risk[i], "추천": int(q[i]), "결정": dec, "확정": final, "사유": why})
|
|
res = pd.DataFrame(rows).sort_values("위험", ascending=False)
|
|
print(res.head(10).to_string(index=False, float_format=lambda x: f"{x:.0f}"))
|
|
print("결정 집계:", ", ".join(f"{k} {v}건" for k, v in res["결정"].value_counts().items()))
|
|
auto = [e for e in audit.log if e["t"] == "추천" and e["status"] != "결재대기"]
|
|
print(f"자동 확정 {len(auto)}건, 결재 사유 기록률 {res['사유'].astype(bool).mean():.0%}, 감사 로그 {len(audit.log)}건 무결성: {'정상' if audit.verify() is None else '변조'}")
|
|
j = next(e["seq"] for e in audit.log if e["t"] == "결재" and e["decision"] != "승인")
|
|
old = audit.log[j]["decision"]; audit.log[j]["decision"] = "승인"
|
|
print(f"변조 시험({j}번 {old} 기록을 승인으로 수정) -> {audit.verify()}번 기록에서 변조 검출")
|
|
|
|
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
|
|
tab[["보관비", "발주비", "결품손실"]].plot.bar(stacked=True, ax=ax[0], rot=0)
|
|
ax[0].set_xticklabels(["current", "forecast", "forecast+EOL alt"]); ax[0].set_title("12-month cost (million, synthetic)"); ax[0].legend(["holding", "ordering", "shortage"])
|
|
sd = pd.DataFrame(sens)
|
|
for nm, g in sd.groupby("수요"):
|
|
ax[1].plot(g["z"], g["총비용"], "o-", label="normal" if nm == "평시" else "surge x1.5")
|
|
ax[1].set_xlabel("service level z"); ax[1].set_ylabel("total cost (million)"); ax[1].set_title("Sensitivity"); ax[1].legend()
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[12개월 재고 백테스트(가상 비용 지수, 백만 원)]
|
|
보관비 발주비 결품손실 총비용(백만) 충족률 회전율
|
|
현행(평균+3개월 여유) 81.6 21.7 234.3 337.6 86.9% 5.42
|
|
예측 정책(z=1.65) 77.2 12.0 235.2 324.4 86.8% 6.09
|
|
예측+단종 대체품 전환 108.9 12.7 8.3 129.8 98.6% 5.65
|
|
총비용 절감 61.5%, 예측 정책만의 회전율 향상 12.2% (목표: 비용 15% 절감 또는 회전율 20% 향상 -> 달성)
|
|
효과 분해: 생산중·단종예고 품목 비용 13.0% 절감(예측 정책), 나머지는 단종 품목 대체품 전환 효과
|
|
[민감도] z=1.28 평시: 125백만, 충족 98.5% / z=1.28 돌발 1.5배: 230백만, 충족 90.4% / z=1.65 평시: 130백만, 충족 98.6% / z=1.65 돌발 1.5배: 213백만, 충족 91.6% / z=2.05 평시: 136백만, 충족 98.7% / z=2.05 돌발 1.5배: 195백만, 충족 92.8%
|
|
|
|
[다음 달 발주 추천 LP] 부족 품목 18개, 예산 93.6백만(부족분 전액의 60%, 가상) 사용 93.5백만, 추천 16건
|
|
품목 구분 위험 추천 결정 확정 사유
|
|
P004 정규 발주 88 45 조정 31 고액 발주 분할(1차 70%), 다음 달 재검토
|
|
P019 정규 발주 88 62 승인 62 위험 지수·재고 근거 확인
|
|
P030 정규 발주 88 491 승인 491 위험 지수·재고 근거 확인
|
|
P025 정규 발주 88 116 조정 81 고액 발주 분할(1차 70%), 다음 달 재검토
|
|
P038 정규 발주 88 57 승인 57 위험 지수·재고 근거 확인
|
|
P027 정규 발주 87 29 조정 20 고액 발주 분할(1차 70%), 다음 달 재검토
|
|
P032 정규 발주 73 26 승인 26 위험 지수·재고 근거 확인
|
|
P006 정규 발주 64 23 승인 23 위험 지수·재고 근거 확인
|
|
P012 정규 발주 61 45 승인 45 위험 지수·재고 근거 확인
|
|
P037 정규 발주 41 14 승인 14 위험 지수·재고 근거 확인
|
|
결정 집계: 승인 11건, 조정 3건, 보류 2건
|
|
자동 확정 0건, 결재 사유 기록률 100%, 감사 로그 32건 무결성: 정상
|
|
변조 시험(1번 조정 기록을 승인으로 수정) -> 1번 기록에서 변조 검출
|
|
```
|
|
|
|
## 미션 8. 전시 작전 템포 증가 모의 부품 수요 300% 폭증 시뮬레이션 부하 테스트
|
|
|
|
- 난이도: 고급 / 예상 소요시간: 3.0시간
|
|
- 코드 설명: 전시 작전 템포 증가로 부품 수요가 평시의 300%로 폭증하는 3개월 상황을 가상으로 모의해, 품목별 결품 경보(다음 달 예측 수요 > 가용재고)의 정확도를 반복 시험하고, 품목 수 50배 부하에서 재고 대시보드 갱신 시간(2초 이내)을 측정한다. 트리 모델이 학습 범위 밖 수요를 외삽하지 못하는 한계를 작전 템포 계수 보정으로 보완한다.
|
|
|
|
```python
|
|
"""전시 작전 템포 증가로 부품 수요가 평시의 300%로 폭증하는 3개월 상황을 가상으로 모의해, 품목별 결품 경보(다음 달 예측 수요 > 가용재고)의 정확도를 반복 시험하고, 품목 수 50배 부하에서 재고 대시보드 갱신 시간(2초 이내)을 측정한다. 트리 모델이 학습 범위 밖 수요를 외삽하지 못하는 한계를 작전 템포 계수 보정으로 보완한다."""
|
|
import time
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy.stats import norm
|
|
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
# ---- 앙상블 수요 예측·안전재고·결품(단종) 위험 지수 공통 함수 ----
|
|
STATUS_W = {"생산중": 0.0, "단종예고": 0.5, "단종": 1.0} # [교육용가상자료] 단종 상태 가중치
|
|
|
|
|
|
def fit_ens(tr):
|
|
"""포아송 그래디언트 부스팅(70%) + 랜덤포레스트(30%) 앙상블."""
|
|
h = new_model().fit(tr[FEATS], tr["demand"])
|
|
r = RandomForestRegressor(n_estimators=150, min_samples_leaf=5, random_state=0, n_jobs=1).fit(tr[FEATS], tr["demand"])
|
|
return h, r
|
|
|
|
|
|
def pred_ens(m, X, w=0.7):
|
|
return np.clip(w * m[0].predict(X[FEATS]) + (1 - w) * m[1].predict(X[FEATS]), 0, None)
|
|
|
|
|
|
def next_month(hist, items, t_new=48, op_mult=1.0, ex=0):
|
|
"""다음 달(t_new) 예측용 행: 작년 같은 달 운용시간 x 배수, 훈련 여부 지정."""
|
|
last = hist[hist["t"] == t_new - 1].copy()
|
|
prev = hist[hist["t"] == t_new - 12].set_index("item_id")["op_hours"]
|
|
m = (t_new % 12) + 1
|
|
f = last.assign(t=t_new, ym=f"{2022 + t_new // 12}-{m:02d}", month=m, demand=0, exercise=ex,
|
|
op_hours=last["item_id"].map(prev).to_numpy() * op_mult)
|
|
X = make_features(pd.concat([hist, f], ignore_index=True))
|
|
return X[X["t"] == t_new].set_index("item_id").loc[items["item_id"]].reset_index()
|
|
|
|
|
|
def stock_policy(items, mu, sigma, z=1.65, review=1):
|
|
"""안전재고 = z x 월 예측오차 표준편차 x sqrt(리드타임+검토주기), 재주문점 = 월 수요 x (리드타임+검토주기) + 안전재고."""
|
|
L = items["lead_time"].to_numpy() + review
|
|
ss = z * sigma * np.sqrt(L)
|
|
return pd.DataFrame({"item_id": items["item_id"], "mu": mu, "sigma": sigma, "L": L,
|
|
"ss": np.ceil(ss), "rop": np.ceil(mu * L + ss)})
|
|
|
|
|
|
def risk_index(items, pol, stock):
|
|
"""결품 위험 지수(0~100): 리드타임 동안 수요가 가용 재고를 넘을 확률(정규 근사)을 기본으로,
|
|
단종 상태·단일 공급업체·대체품 부재에 가중을 준다(가상 가중치)."""
|
|
st = stock.set_index("item_id").loc[items["item_id"]]
|
|
avail = (st["on_hand"] + st["on_order"]).to_numpy()
|
|
p = 1 - norm.cdf((avail - pol["mu"].to_numpy() * pol["L"].to_numpy()) / (pol["sigma"].to_numpy() * np.sqrt(pol["L"].to_numpy()) + 1e-6))
|
|
sw = items["status"].map(STATUS_W).to_numpy()
|
|
sup = (items["suppliers"].to_numpy() == 1).astype(float)
|
|
alt = (~items["alt_available"].to_numpy()).astype(float)
|
|
score = 100 * np.clip(0.55 * p + 0.25 * sw + 0.10 * sup + 0.10 * alt, 0, 1)
|
|
return pd.DataFrame({"item_id": items["item_id"], "p_short": p, "status_w": sw, "single_sup": sup, "no_alt": alt,
|
|
"avail": avail, "risk": score.round(1)})
|
|
|
|
SURGE = 3.0 # 계획상 작전 템포 수요 배수(300%)
|
|
GROUP_MULT = {"power_train": 3.6, "brake_steer": 3.3, "fuel": 3.0, "cooling": 2.8, # [교육용가상자료] 실제 배수(계획에는 미반영)
|
|
"filter": 2.7, "gasket": 3.2, "battery": 2.6, "semiconductor": 3.1}
|
|
REPS, MONTHS = 30, 3
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
df = make_features(hist)
|
|
stock = make_stock(items, hist)
|
|
ens = fit_ens(df)
|
|
X = next_month(hist, items, op_mult=SURGE, ex=1)
|
|
raw = pred_ens(ens, X) # 운용시간 3배를 넣은 모델 예측(학습 범위 밖)
|
|
base = pred_ens(ens, next_month(hist, items)) # 평시 예측
|
|
adj = base * SURGE # 작전 템포 계수 보정 예측
|
|
print(f"[폭증 예측] 평시 합계 {base.sum():.0f}개, 모델 직접 예측 {raw.sum():.0f}개({raw.sum() / base.sum():.1f}배), 템포 계수 보정 {adj.sum():.0f}개({SURGE:.0f}배)")
|
|
print("-> 트리 모델은 학습한 운용시간 범위를 넘으면 예측이 더 늘지 않는다(외삽 불가). 계획 배수로 보정한다.")
|
|
|
|
|
|
def run(fc, seed, adaptive=False):
|
|
"""3개월 모의: 매월 경보(예측 수요 > 월초 재고) vs 실제 결품(실제 수요 > 월초 재고).
|
|
adaptive=True이면 지난달 품목군별 실제/평시 비율로 다음 달 배수를 갱신한다(실적 반영)."""
|
|
rng = np.random.default_rng(seed)
|
|
true_mult = items["group"].map(GROUP_MULT).to_numpy() * rng.lognormal(0, 0.12, len(items))
|
|
on = stock["on_hand"].to_numpy().astype(float)
|
|
arrive = stock["on_order"].to_numpy() / np.maximum(items["lead_time"].to_numpy(), 1)
|
|
al, tr, bd = [], [], []
|
|
grp = items["group"].to_numpy(); f = fc.copy()
|
|
for _ in range(MONTHS):
|
|
actual = rng.poisson(base * true_mult)
|
|
al.append(f > on); tr.append(actual > on); bd.append(np.abs(f - on) <= 1.65 * np.sqrt(np.maximum(f, 1)))
|
|
if adaptive:
|
|
ratio = pd.Series(actual).groupby(grp).sum() / pd.Series(base).groupby(grp).sum()
|
|
f = base * pd.Series(grp).map(ratio).to_numpy()
|
|
on = np.maximum(on - actual, 0) + arrive
|
|
return np.array(al), np.array(tr), np.array(bd)
|
|
|
|
|
|
rows = []
|
|
for nm, fc, ad in (("평시 예측(보정 없음)", base, False), ("모델 직접 예측", raw, False), ("템포 계수 보정", adj, False),
|
|
("템포 보정+월별 실적 반영", adj, True)):
|
|
acc, rec, n_al = [], [], []
|
|
for s in range(REPS):
|
|
a, t, bd = run(fc, s, ad)
|
|
acc.append((a == t).mean()); rec.append((a & t).sum() / max(t.sum(), 1)); n_al.append(a.sum())
|
|
if ad:
|
|
clear = [(lambda a, t, bd: ((a == t)[~bd].mean(), bd.mean()))(*run(fc, s, ad)) for s in range(REPS)]
|
|
rows.append({"예측 방식": nm, "경보 정확도": np.mean(acc), "최저 정확도": np.min(acc), "결품 탐지율": np.mean(rec), "월평균 경보": np.mean(n_al) / MONTHS})
|
|
res = pd.DataFrame(rows)
|
|
print(f"\n[경보 시험: 품목 {len(items)}개 x {MONTHS}개월 x 반복 {REPS}회]")
|
|
print(res.to_string(index=False, formatters={"경보 정확도": "{:.1%}".format, "최저 정확도": "{:.1%}".format, "결품 탐지율": "{:.1%}".format, "월평균 경보": "{:.1f}".format}))
|
|
acc = res.iloc[3]["경보 정확도"]
|
|
print(f"템포 보정+실적 반영 경보 정확도 {acc:.1%} (목표 95% 이상: {'달성' if acc >= 0.95 else '미달'}), 템포 보정만 {res.iloc[2]['경보 정확도']:.1%}")
|
|
print(f"오경보 원인: 예측과 재고 차이가 수요 변동 폭(1.65 x 표준편차) 이내인 경계 구간이 판정의 {np.mean([c[1] for c in clear]):.0%}를 차지한다. "
|
|
f"경계 구간을 '주의(검토 필요)'로 따로 표시하면 나머지 판정 정확도는 {np.mean([c[0] for c in clear]):.1%}이다.")
|
|
|
|
|
|
# 부하 시험: 품목 50배(2,000개) 대시보드 갱신 = 예측 + 위험 지수 + 집계
|
|
def refresh(Xb, it, st):
|
|
mu = pred_ens(ens, Xb) * SURGE
|
|
pol = stock_policy(it, mu, np.sqrt(np.maximum(mu, 1)))
|
|
rk = risk_index(it, pol, st)
|
|
return rk.assign(group=it["group"].to_numpy()).groupby("group")["risk"].agg(["mean", "max"]), rk
|
|
|
|
|
|
K = 50
|
|
Xb = pd.concat([X] * K, ignore_index=True)
|
|
itb = pd.concat([items.assign(item_id=items["item_id"] + f"_{k}") for k in range(K)], ignore_index=True)
|
|
stb = pd.concat([stock.assign(item_id=stock["item_id"] + f"_{k}") for k in range(K)], ignore_index=True)
|
|
times = []
|
|
for _ in range(3):
|
|
t0 = time.perf_counter(); agg, rk = refresh(Xb, itb, stb); times.append(time.perf_counter() - t0)
|
|
t1 = time.perf_counter(); refresh(X, items, stock); t_small = time.perf_counter() - t1
|
|
print(f"\n[대시보드 부하 시험] 품목 {len(itb)}개 갱신 평균 {np.mean(times):.2f}s / 최대 {max(times):.2f}s, 품목 40개 {t_small:.3f}s (목표 2초 이내: {'달성' if max(times) <= 2 else '미달'})")
|
|
print("폭증 시 품목군별 최대 위험 지수:", ", ".join(f"{g} {r['max']:.0f}" for g, r in agg.iterrows()))
|
|
print("경보는 조달 검토 요청이며 긴급 조달·대체품 전환은 정비장교 승인 후 진행한다.")
|
|
|
|
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
|
|
ax[0].bar(["normal", "raw", "tempo", "tempo+AAR"], res["경보 정확도"], color=["gray", "tab:orange", "tab:green", "tab:blue"])
|
|
ax[0].axhline(0.95, c="r", ls="--"); ax[0].set_ylim(0.5, 1); ax[0].set_title("Alert accuracy under 300% demand surge")
|
|
ax[1].scatter(base, raw, s=12, label="model raw"); ax[1].scatter(base, adj, s=12, label="tempo adjusted")
|
|
ax[1].plot([0, base.max()], [0, base.max() * 3], "k--", lw=1, label="x3 line"); ax[1].set_xlabel("normal forecast"); ax[1].set_ylabel("surge forecast")
|
|
ax[1].set_title("Tree model cannot extrapolate"); ax[1].legend()
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[폭증 예측] 평시 합계 948개, 모델 직접 예측 1193개(1.3배), 템포 계수 보정 2843개(3배)
|
|
-> 트리 모델은 학습한 운용시간 범위를 넘으면 예측이 더 늘지 않는다(외삽 불가). 계획 배수로 보정한다.
|
|
|
|
[경보 시험: 품목 40개 x 3개월 x 반복 30회]
|
|
예측 방식 경보 정확도 최저 정확도 결품 탐지율 월평균 경보
|
|
평시 예측(보정 없음) 45.9% 40.8% 40.3% 14.6
|
|
모델 직접 예측 65.0% 61.7% 61.5% 22.3
|
|
템포 계수 보정 93.6% 88.3% 98.5% 37.7
|
|
템포 보정+월별 실적 반영 93.8% 88.3% 98.4% 37.5
|
|
템포 보정+실적 반영 경보 정확도 93.8% (목표 95% 이상: 미달), 템포 보정만 93.6%
|
|
오경보 원인: 예측과 재고 차이가 수요 변동 폭(1.65 x 표준편차) 이내인 경계 구간이 판정의 19%를 차지한다. 경계 구간을 '주의(검토 필요)'로 따로 표시하면 나머지 판정 정확도는 99.6%이다.
|
|
|
|
[대시보드 부하 시험] 품목 2000개 갱신 평균 0.05s / 최대 0.05s, 품목 40개 0.024s (목표 2초 이내: 달성)
|
|
폭증 시 품목군별 최대 위험 지수: battery 100, brake_steer 88, cooling 65, filter 88, fuel 88, gasket 90, power_train 90, semiconductor 78
|
|
경보는 조달 검토 요청이며 긴급 조달·대체품 전환은 정비장교 승인 후 진행한다.
|
|
```
|
|
|
|
## 미션 9. AAR 기반 부품 과다/과소 재고 예측 오차 분석 및 안전재고 파라미터 재보정
|
|
|
|
- 난이도: 고급 / 예상 소요시간: 3.0시간
|
|
- 코드 설명: AAR 관점에서 지난 12개월 재고 운영의 과다·과소 재고와 수요 예측 오차를 품목군·생산 상태·ABC 등급별로 분석해 원인을 찾고, 앞 6개월 자료로 ABC 등급별 안전재고 계수(z)와 품목군별 예측 편향 보정값을 재보정한 뒤 뒤 6개월에서 검증하며, 비용 절감액·정비 대기시간 단축률·재고 회전율 3대 효과 지표와 민감도, 도입 로드맵을 산출한다.
|
|
|
|
```python
|
|
"""AAR 관점에서 지난 12개월 재고 운영의 과다·과소 재고와 수요 예측 오차를 품목군·생산 상태·ABC 등급별로 분석해 원인을 찾고, 앞 6개월 자료로 ABC 등급별 안전재고 계수(z)와 품목군별 예측 편향 보정값을 재보정한 뒤 뒤 6개월에서 검증하며, 비용 절감액·정비 대기시간 단축률·재고 회전율 3대 효과 지표와 민감도, 도입 로드맵을 산출한다."""
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy.stats import norm
|
|
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
# ---- 앙상블 수요 예측·안전재고·결품(단종) 위험 지수 공통 함수 ----
|
|
STATUS_W = {"생산중": 0.0, "단종예고": 0.5, "단종": 1.0} # [교육용가상자료] 단종 상태 가중치
|
|
|
|
|
|
def fit_ens(tr):
|
|
"""포아송 그래디언트 부스팅(70%) + 랜덤포레스트(30%) 앙상블."""
|
|
h = new_model().fit(tr[FEATS], tr["demand"])
|
|
r = RandomForestRegressor(n_estimators=150, min_samples_leaf=5, random_state=0, n_jobs=1).fit(tr[FEATS], tr["demand"])
|
|
return h, r
|
|
|
|
|
|
def pred_ens(m, X, w=0.7):
|
|
return np.clip(w * m[0].predict(X[FEATS]) + (1 - w) * m[1].predict(X[FEATS]), 0, None)
|
|
|
|
|
|
def next_month(hist, items, t_new=48, op_mult=1.0, ex=0):
|
|
"""다음 달(t_new) 예측용 행: 작년 같은 달 운용시간 x 배수, 훈련 여부 지정."""
|
|
last = hist[hist["t"] == t_new - 1].copy()
|
|
prev = hist[hist["t"] == t_new - 12].set_index("item_id")["op_hours"]
|
|
m = (t_new % 12) + 1
|
|
f = last.assign(t=t_new, ym=f"{2022 + t_new // 12}-{m:02d}", month=m, demand=0, exercise=ex,
|
|
op_hours=last["item_id"].map(prev).to_numpy() * op_mult)
|
|
X = make_features(pd.concat([hist, f], ignore_index=True))
|
|
return X[X["t"] == t_new].set_index("item_id").loc[items["item_id"]].reset_index()
|
|
|
|
|
|
def stock_policy(items, mu, sigma, z=1.65, review=1):
|
|
"""안전재고 = z x 월 예측오차 표준편차 x sqrt(리드타임+검토주기), 재주문점 = 월 수요 x (리드타임+검토주기) + 안전재고."""
|
|
L = items["lead_time"].to_numpy() + review
|
|
ss = z * sigma * np.sqrt(L)
|
|
return pd.DataFrame({"item_id": items["item_id"], "mu": mu, "sigma": sigma, "L": L,
|
|
"ss": np.ceil(ss), "rop": np.ceil(mu * L + ss)})
|
|
|
|
|
|
def risk_index(items, pol, stock):
|
|
"""결품 위험 지수(0~100): 리드타임 동안 수요가 가용 재고를 넘을 확률(정규 근사)을 기본으로,
|
|
단종 상태·단일 공급업체·대체품 부재에 가중을 준다(가상 가중치)."""
|
|
st = stock.set_index("item_id").loc[items["item_id"]]
|
|
avail = (st["on_hand"] + st["on_order"]).to_numpy()
|
|
p = 1 - norm.cdf((avail - pol["mu"].to_numpy() * pol["L"].to_numpy()) / (pol["sigma"].to_numpy() * np.sqrt(pol["L"].to_numpy()) + 1e-6))
|
|
sw = items["status"].map(STATUS_W).to_numpy()
|
|
sup = (items["suppliers"].to_numpy() == 1).astype(float)
|
|
alt = (~items["alt_available"].to_numpy()).astype(float)
|
|
score = 100 * np.clip(0.55 * p + 0.25 * sw + 0.10 * sup + 0.10 * alt, 0, 1)
|
|
return pd.DataFrame({"item_id": items["item_id"], "p_short": p, "status_w": sw, "single_sup": sup, "no_alt": alt,
|
|
"avail": avail, "risk": score.round(1)})
|
|
|
|
# ---- 월 단위 재고 백테스트(가상 비용 지수) ----
|
|
HOLD, ORDER_COST, SHORT_MULT = 0.02, 50_000, 0.6 # [교육용가상자료] 월 보관비율(단가 대비), 발주 1회 비용(원), 결품 1개 손실(단가 배수)
|
|
|
|
|
|
ALT_PREMIUM, ALT_LT = 1.2, 3 # [교육용가상자료] 대체품 전환 시 단가 배수, 대체품 조달 리드타임(월)
|
|
|
|
|
|
def simulate(items, demand, fc, policy, z=1.65, months=None, sigma=None, eol=True):
|
|
"""품목별 월 재고 모의. demand/fc: (품목 x 월) 실제·예측 수요. policy: 'current'(최근 12개월 평균 + 3개월 여유 고정) 또는 'ai'(예측 + 안전재고).
|
|
단종 품목은 원 품목 발주 불가. 'ai' 정책에서 eol=True이면 단종 품목 중 대체품이 있는 품목은 대체품(단가 x1.2, 리드타임 3개월)으로 전환 발주한다. 반환: 품목별 보관비·발주비·결품손실·불출량·평균재고."""
|
|
n, T = demand.shape
|
|
L = items["lead_time"].to_numpy(); price = items["price"].to_numpy().astype(float)
|
|
can_order = (items["status"] != "단종").to_numpy()
|
|
if policy == "ai" and eol:
|
|
sw = (~can_order) & items["alt_available"].to_numpy()
|
|
can_order = can_order | sw
|
|
L = np.where(sw, ALT_LT, L); price = np.where(sw, price * ALT_PREMIUM, price)
|
|
avg = demand[:, :1].repeat(T, 1) if months is None else months
|
|
on = np.round(avg[:, 0] * 2.0)
|
|
pipe = np.zeros((n, T + int(L.max()) + 1)) # 도착 예정 물량(품목 x 월)
|
|
for i in range(n):
|
|
pipe[i, :L[i]] = np.round(avg[i, 0] * (1.0 if can_order[i] else 0.3))
|
|
res = np.zeros((n, 6)); rows = np.arange(n)
|
|
for t in range(T):
|
|
on += pipe[:, t]
|
|
issue = np.minimum(on, demand[:, t]); short = demand[:, t] - issue; on -= issue
|
|
ip = on + pipe[:, t + 1:].sum(1)
|
|
if policy == "current":
|
|
target = avg[:, t] * (L + 3)
|
|
else:
|
|
s = sigma if sigma is not None else np.sqrt(np.maximum(fc[:, t], 1))
|
|
target = fc[:, t] * (L + 1) + z * s * np.sqrt(L + 1)
|
|
q = np.where(can_order & (ip < target), np.ceil(target - ip), 0)
|
|
pipe[rows, t + L] += q
|
|
res += np.c_[HOLD * price * on, ORDER_COST * (q > 0), SHORT_MULT * price * short, issue, on, short]
|
|
out = pd.DataFrame(res, columns=["hold", "order", "short", "issued", "inv", "short_qty"], index=items["item_id"])
|
|
out["inv"] /= T
|
|
return out
|
|
|
|
WAIT_H = 36 # [교육용가상자료] 결품 1개당 장비 정비 대기시간(시간)
|
|
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
df = make_features(hist)
|
|
m = fit_ens(df[df["t"] < 36])
|
|
bt = df[df["t"] >= 36].copy(); bt["fc"] = pred_ens(m, bt)
|
|
P = lambda c: bt.pivot(index="item_id", columns="t", values=c).loc[items["item_id"]].to_numpy()
|
|
D, F, A = P("demand"), P("fc"), P("item_level")
|
|
sig = np.maximum(0.25 * F.mean(1), np.sqrt(np.maximum(F.mean(1), 1)))
|
|
val = items["price"].to_numpy() * D.sum(1)
|
|
cum = np.cumsum(np.sort(val)[::-1]) / val.sum()
|
|
abc = pd.Series(np.select([val >= np.sort(val)[::-1][np.searchsorted(cum, 0.7)], val >= np.sort(val)[::-1][np.searchsorted(cum, 0.9)]], ["A", "B"], "C"), index=items["item_id"])
|
|
|
|
# 1) AAR 편차 분석: 예측 편향과 과다·과소 재고
|
|
an = items[["item_id", "group", "status"]].assign(abc=abc.values, bias=(F[:, :6].sum(1) - D[:, :6].sum(1)) / np.maximum(D[:, :6].sum(1), 1))
|
|
r0 = simulate(items, D[:, :6], F[:, :6], "ai", 1.65, A[:, :6], sig)
|
|
an["over"] = (r0["inv"].to_numpy() > 3 * np.maximum(D[:, :6].mean(1), 1)) # 평균 재고가 3개월 수요 초과 = 과다
|
|
an["under"] = r0["short_qty"].to_numpy() > 0
|
|
print("[AAR 편차 분석: 앞 6개월(t=36~41)]")
|
|
g = an.groupby("group").agg(예측편향=("bias", "mean"), 과다품목=("over", "sum"), 결품품목=("under", "sum"))
|
|
gt = g.T.astype(object)
|
|
gt.loc["예측편향"] = [f"{v:+.0%}" for v in g["예측편향"]]
|
|
gt.loc[["과다품목", "결품품목"]] = gt.loc[["과다품목", "결품품목"]].map(lambda v: f"{int(v)}")
|
|
print(gt.to_string())
|
|
print("ABC별 과다/결품 품목:", ", ".join(f"{k} {int(v['over'])}/{int(v['under'])}" for k, v in an.groupby("abc")[["over", "under"]].sum().iterrows()),
|
|
"| 생산 상태별 결품 품목:", ", ".join(f"{k} {int(v)}" for k, v in an.groupby("status")["under"].sum().items()))
|
|
ex = bt["exercise"] == 1
|
|
hi = g["예측편향"].abs().sort_values(ascending=False).index[:2]
|
|
hi_txt = ", ".join(f"{h}({g.loc[h, '예측편향']:+.0%})" for h in hi)
|
|
print(f"훈련 월 예측 편향 {(bt.loc[ex, 'fc'].sum() / bt.loc[ex, 'demand'].sum() - 1):+.1%}, 평시 월 {(bt.loc[~ex, 'fc'].sum() / bt.loc[~ex, 'demand'].sum() - 1):+.1%}"
|
|
f" -> 원인: {hi_txt} 품목군 예측 편향, 모든 등급에 같은 z=1.65 적용, 단종 품목 발주 불가")
|
|
|
|
# 2) 재보정(앞 6개월로 결정): ABC별 z 격자 탐색 + 품목군별 편향 보정
|
|
grp_bias = an.groupby("group")["bias"].median().clip(-0.3, 0.3)
|
|
corr = 1 / (1 + items["group"].map(grp_bias).to_numpy())
|
|
best = {}
|
|
for cls in "ABC":
|
|
mask = (abc == cls).to_numpy()
|
|
costs = {}
|
|
for z in (0.84, 1.28, 1.65, 2.05, 2.33):
|
|
zz = np.where(mask, z, 1.65)
|
|
r = simulate(items, D[:, :6], F[:, :6] * corr[:, None], "ai", zz, A[:, :6], sig)
|
|
costs[z] = r[["hold", "order", "short"]].to_numpy()[mask].sum()
|
|
best[cls] = min(costs, key=costs.get)
|
|
z_new = abc.map(best).to_numpy()
|
|
print(f"\n[재보정 결과] ABC별 z: A {best['A']}, B {best['B']}, C {best['C']} (기존 1.65 일괄), 품목군 편향 보정 범위 {corr.min():.2f}~{corr.max():.2f}배")
|
|
|
|
# 3) 검증(뒤 6개월, t=42~47) 및 3대 효과 지표
|
|
def ev(fc, z):
|
|
r = simulate(items, D[:, 6:], fc, "ai", z, A[:, 6:], sig)
|
|
return r, r[["hold", "order", "short"]].sum().sum()
|
|
|
|
|
|
r_cur = simulate(items, D[:, 6:], F[:, 6:], "current", months=A[:, 6:])
|
|
r_old, c_old = ev(F[:, 6:], 1.65)
|
|
r_new, c_new = ev(F[:, 6:] * corr[:, None], z_new)
|
|
c_cur = r_cur[["hold", "order", "short"]].sum().sum()
|
|
turn = lambda r: float(r["issued"].sum() / r["inv"].sum())
|
|
over_amt = lambda r: float((np.maximum(r["inv"].to_numpy() - 3 * np.maximum(D[:, 6:].mean(1), 1), 0) * items["price"].to_numpy()).sum())
|
|
k = pd.DataFrame([["6개월 총비용(백만 원)", c_cur / 1e6, c_old / 1e6, c_new / 1e6],
|
|
["정비 대기시간(시간)", r_cur["short_qty"].sum() * WAIT_H, r_old["short_qty"].sum() * WAIT_H, r_new["short_qty"].sum() * WAIT_H],
|
|
["재고 회전율(회/6개월)", turn(r_cur), turn(r_old), turn(r_new)]],
|
|
columns=["지표", "현행 정책", "예측 정책(기존 파라미터)", "예측 정책(재보정)"]).set_index("지표")
|
|
k["현행 대비 개선율"] = 1 - k["예측 정책(재보정)"] / k["현행 정책"]
|
|
k.loc["재고 회전율(회/6개월)", "현행 대비 개선율"] *= -1 # 회전율은 클수록 좋음
|
|
print("\n[3대 정량 효과 지표: 뒤 6개월 검증, 가상 단가 기준]")
|
|
print(k.to_string(float_format=lambda x: f"{x:,.1f}", formatters={"현행 대비 개선율": "{:.1%}".format}))
|
|
print(f"재보정 효과(기존 파라미터 대비): 총비용 {1 - c_new / c_old:+.1%}, 연간 환산 절감액(현행 대비) {(c_cur - c_new) * 2 / 1e6:,.0f}백만 원(가상)")
|
|
sens = []
|
|
for hm in (0.5, 1.0, 1.5):
|
|
for sm in (0.3, 0.6, 1.0):
|
|
HOLD_, SM_ = HOLD * hm, sm
|
|
cc = lambda r, p: (HOLD_ * p * r["inv"].to_numpy() * 6).sum() + r["order"].sum() + (SM_ * p * r["short_qty"].to_numpy()).sum()
|
|
pr = items["price"].to_numpy()
|
|
sens.append(f"보관x{hm}/결품x{sm}: {1 - cc(r_new, pr) / cc(r_cur, pr):.0%}")
|
|
print("[민감도: 현행 대비 비용 절감률]", ", ".join(sens))
|
|
print("도입 로드맵: 단기(1~3개월) 분기별 AAR로 z·편향 재보정 / 중기(4~9개월) 단종예고 품목 최종 구매·대체품 검증 절차 정착 / 장기(10개월~) 군수 정보체계 연계와 자동 재학습")
|
|
print("금액·시간은 가정 단가로 계산한 상대 비교값이며 예산 근거로 쓰지 않는다.")
|
|
|
|
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
|
|
g["예측편향"].plot.bar(ax=ax[0], rot=30, color="tab:purple"); ax[0].axhline(0, c="k", lw=0.8); ax[0].set_title("Forecast bias by group (AAR, first 6 months)")
|
|
k.iloc[:, :3].div(k["현행 정책"], axis=0).T.plot.bar(ax=ax[1], rot=0); ax[1].set_xticklabels(["current", "forecast(old)", "forecast(recal)"])
|
|
ax[1].set_title("Effect indicators (current = 1.0)"); ax[1].legend(["cost", "wait hours", "turnover"])
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[AAR 편차 분석: 앞 6개월(t=36~41)]
|
|
group battery brake_steer cooling filter fuel gasket power_train semiconductor
|
|
예측편향 +11% +18% +4% +1% -6% -5% +1% -3%
|
|
과다품목 1 1 1 2 0 0 0 0
|
|
결품품목 1 0 0 0 1 0 0 0
|
|
ABC별 과다/결품 품목: A 1/0, B 2/0, C 2/2 | 생산 상태별 결품 품목: 단종 1, 단종예고 0, 생산중 1
|
|
훈련 월 예측 편향 +1.5%, 평시 월 +0.1% -> 원인: brake_steer(+18%), battery(+11%) 품목군 예측 편향, 모든 등급에 같은 z=1.65 적용, 단종 품목 발주 불가
|
|
|
|
[재보정 결과] ABC별 z: A 0.84, B 0.84, C 1.28 (기존 1.65 일괄), 품목군 편향 보정 범위 0.89~1.07배
|
|
|
|
[3대 정량 효과 지표: 뒤 6개월 검증, 가상 단가 기준]
|
|
현행 정책 예측 정책(기존 파라미터) 예측 정책(재보정) 현행 대비 개선율
|
|
지표
|
|
6개월 총비용(백만 원) 117.7 56.7 55.3 53.0%
|
|
정비 대기시간(시간) 17,532.0 1,620.0 1,872.0 89.3%
|
|
재고 회전율(회/6개월) 3.1 3.0 3.1 1.2%
|
|
재보정 효과(기존 파라미터 대비): 총비용 +2.5%, 연간 환산 절감액(현행 대비) 125백만 원(가상)
|
|
[민감도: 현행 대비 비용 절감률] 보관x0.5/결품x0.3: 55%, 보관x0.5/결품x0.6: 69%, 보관x0.5/결품x1.0: 77%, 보관x1.0/결품x0.3: 38%, 보관x1.0/결품x0.6: 55%, 보관x1.0/결품x1.0: 67%, 보관x1.5/결품x0.3: 28%, 보관x1.5/결품x0.6: 45%, 보관x1.5/결품x1.0: 58%
|
|
도입 로드맵: 단기(1~3개월) 분기별 AAR로 z·편향 재보정 / 중기(4~9개월) 단종예고 품목 최종 구매·대체품 검증 절차 정착 / 장기(10개월~) 군수 정보체계 연계와 자동 재학습
|
|
금액·시간은 가정 단가로 계산한 상대 비교값이며 예산 근거로 쓰지 않는다.
|
|
```
|
|
|
|
## 미션 10. AI 기반 야전 정비 부품 수요예측 및 재고최적화 시스템 통합 패키지 완성
|
|
|
|
- 난이도: 최고급 / 예상 소요시간: 3.5시간
|
|
- 코드 설명: 수리부속 청구 이력 생성·정합성 점검, 수요 예측 모델 자동 재학습(기존·신규 모델 비교 후 교체), 안전재고·결품(단종) 위험 지수, 단종예고 품목 최종 구매량과 단종 품목 대체 검토, 예산 제약 발주 추천과 정비장교 결재 대기열을 한 번에 실행하는 통합 패키지를 완성하고, 자체 점검표·운영 가이드 목차·체계 연계 메시지 예시·브리핑 요약을 출력한다.
|
|
|
|
```python
|
|
"""수리부속 청구 이력 생성·정합성 점검, 수요 예측 모델 자동 재학습(기존·신규 모델 비교 후 교체), 안전재고·결품(단종) 위험 지수, 단종예고 품목 최종 구매량과 단종 품목 대체 검토, 예산 제약 발주 추천과 정비장교 결재 대기열을 한 번에 실행하는 통합 패키지를 완성하고, 자체 점검표·운영 가이드 목차·체계 연계 메시지 예시·브리핑 요약을 출력한다."""
|
|
import json
|
|
import time
|
|
import numpy as np
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from scipy.stats import norm
|
|
from scipy.optimize import linprog
|
|
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
|
|
from sklearn.metrics import mean_absolute_error
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# [교육용가상자료] 가상 야전 정비창 수리부속 청구/출고·장비 가동률 데이터 생성기
|
|
# 실제 부대·품목·재고·단가 정보와 무관하다. 품목 40개 x 48개월(가상 2022-01 ~ 2025-12) 월별 자료를 만든다.
|
|
# ---------------------------------------------------------------------------
|
|
FSC = {"2520": "power_train", "2530": "brake_steer", "2910": "fuel", "2930": "cooling",
|
|
"2940": "filter", "5330": "gasket", "6140": "battery", "5961": "semiconductor"}
|
|
EXERCISE = {3, 8, 10} # 가상 대규모 훈련 월(훈련 계획으로 미리 알 수 있음)
|
|
N_MONTHS = 48
|
|
|
|
|
|
def season(fsc, month):
|
|
"""품목군별 계절 계수: 축전지는 겨울, 냉각계통은 여름에 수요가 늘어난다."""
|
|
if fsc == "6140":
|
|
return 1.6 if month in (12, 1, 2) else 0.9
|
|
if fsc == "2930":
|
|
return 1.5 if month in (6, 7, 8) else 0.9
|
|
return 1 + 0.1 * np.sin(2 * np.pi * month / 12)
|
|
|
|
|
|
def make_items(n=40, seed=2):
|
|
"""품목 마스터: NSN(가상), 품목군, 장비 유형, 기준 수요, 단가, 조달 리드타임, 공급업체 수, 생산(단종) 상태."""
|
|
rng = np.random.default_rng(seed)
|
|
rows = []
|
|
for i in range(n):
|
|
f = str(rng.choice(list(FSC)))
|
|
status = str(rng.choice(["생산중", "단종예고", "단종"], p=[0.7, 0.2, 0.1]))
|
|
rows.append({"item_id": f"P{i + 1:03d}", "nsn": f"{f}-00-{rng.integers(100, 1000)}-{rng.integers(1000, 10000)}",
|
|
"fsc": f, "group": FSC[f], "equip_type": str(rng.choice(["A", "B", "C"])),
|
|
"base": float(np.round(rng.lognormal(2.7, 0.8), 1)),
|
|
"price": int(rng.lognormal(12, 1.0)) // 100 * 100,
|
|
"lead_time": int(rng.integers(1, 7) if status == "생산중" else rng.integers(4, 13)),
|
|
"suppliers": int(rng.integers(1, 5)) if status == "생산중" else 1,
|
|
"status": status, "alt_available": bool(rng.random() < (0.6 if status == "생산중" else 0.3))})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def make_history(items, months=N_MONTHS, seed=3):
|
|
"""월별 청구 수요(개), 긴급 불출 건수, 장비 유형별 운용시간·가동률."""
|
|
rng = np.random.default_rng(seed)
|
|
t = np.arange(months)
|
|
moy = t % 12 + 1
|
|
ex = np.isin(moy, list(EXERCISE))
|
|
hours = {e: (100 + 20 * rng.random(months)) * np.where(ex, 1.5, 1.0) for e in "ABC"}
|
|
avail = {e: np.clip(0.93 - 0.12 * (hours[e] - 100) / 80 + rng.normal(0, 0.015, months), 0.6, 0.99) for e in "ABC"}
|
|
rows = []
|
|
for it in items.itertuples():
|
|
trend = 1 + rng.uniform(-0.1, 0.3) * t / months
|
|
s = np.array([season(it.fsc, m) for m in moy])
|
|
lam = it.base * s * trend * hours[it.equip_type] / 110
|
|
d = rng.poisson(lam)
|
|
rows.append(pd.DataFrame({"item_id": it.item_id, "t": t, "ym": [f"{2022 + k // 12}-{k % 12 + 1:02d}" for k in t],
|
|
"month": moy, "exercise": ex.astype(int), "op_hours": hours[it.equip_type].round(1),
|
|
"availability": avail[it.equip_type].round(3), "demand": d,
|
|
"emergency": rng.binomial(d, 0.05)}))
|
|
return pd.concat(rows, ignore_index=True)
|
|
|
|
|
|
def make_stock(items, hist, seed=4):
|
|
"""현재(가상 2026-01) 재고: 보유량과 미입고 발주량(조달 중인 물량). 단종 품목은 추가 발주가 어려워 미입고량이 적다."""
|
|
rng = np.random.default_rng(seed)
|
|
recent = hist[hist["t"] >= hist["t"].max() - 5].groupby("item_id")["demand"].mean()
|
|
st = items[["item_id", "lead_time", "status"]].copy()
|
|
st["on_hand"] = [int(round(recent[i] * rng.uniform(0.5, 3.0))) for i in st["item_id"]]
|
|
cover = np.where(st["status"] == "생산중", rng.uniform(0.6, 1.3, len(st)), rng.uniform(0.0, 0.6, len(st)))
|
|
st["on_order"] = [int(round(recent[i] * L * c)) for i, L, c in zip(st["item_id"], st["lead_time"], cover)]
|
|
return st[["item_id", "on_hand", "on_order"]]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---- 수요 예측 공통 함수(파생 피처 + 머신러닝 모델) ----
|
|
FEATS = ["lag1", "lag2", "lag3", "lag12", "ma3", "ma6", "std6", "season_idx", "month_sin", "month_cos",
|
|
"exercise", "op_hours", "op_ratio", "item_level", "trend_t"]
|
|
|
|
|
|
def make_features(hist):
|
|
"""품목별 과거 수요로 파생 피처를 만든다(모두 예측 시점 이전 정보만 사용)."""
|
|
df = hist.sort_values(["item_id", "t"]).reset_index(drop=True)
|
|
g = df.groupby("item_id")["demand"]
|
|
for k in (1, 2, 3, 12):
|
|
df[f"lag{k}"] = g.shift(k)
|
|
df["ma3"] = g.transform(lambda s: s.shift(1).rolling(3).mean())
|
|
df["ma6"] = g.transform(lambda s: s.shift(1).rolling(6).mean())
|
|
df["std6"] = g.transform(lambda s: s.shift(1).rolling(6).std())
|
|
df["item_level"] = g.transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["season_idx"] = df["lag12"] / (g.transform(lambda s: s.shift(12).rolling(12).mean()) + 1)
|
|
df["month_sin"] = np.sin(2 * np.pi * df["month"] / 12)
|
|
df["month_cos"] = np.cos(2 * np.pi * df["month"] / 12)
|
|
df["op_ratio"] = df["op_hours"] / df.groupby("item_id")["op_hours"].transform(lambda s: s.shift(1).rolling(12).mean())
|
|
df["trend_t"] = df["t"]
|
|
return df.dropna().reset_index(drop=True)
|
|
|
|
|
|
def new_model():
|
|
"""Prophet/LSTM 대신 사용하는 포아송 손실 그래디언트 부스팅(개수 자료에 적합)."""
|
|
return HistGradientBoostingRegressor(loss="poisson", max_iter=300, learning_rate=0.05, max_leaf_nodes=15,
|
|
min_samples_leaf=20, random_state=0)
|
|
|
|
|
|
def mape(y, p, min_y=1):
|
|
y, p = np.asarray(y, float), np.asarray(p, float)
|
|
m = y >= min_y
|
|
return float(np.mean(np.abs(y[m] - p[m]) / y[m]))
|
|
|
|
# ---- 앙상블 수요 예측·안전재고·결품(단종) 위험 지수 공통 함수 ----
|
|
STATUS_W = {"생산중": 0.0, "단종예고": 0.5, "단종": 1.0} # [교육용가상자료] 단종 상태 가중치
|
|
|
|
|
|
def fit_ens(tr):
|
|
"""포아송 그래디언트 부스팅(70%) + 랜덤포레스트(30%) 앙상블."""
|
|
h = new_model().fit(tr[FEATS], tr["demand"])
|
|
r = RandomForestRegressor(n_estimators=150, min_samples_leaf=5, random_state=0, n_jobs=1).fit(tr[FEATS], tr["demand"])
|
|
return h, r
|
|
|
|
|
|
def pred_ens(m, X, w=0.7):
|
|
return np.clip(w * m[0].predict(X[FEATS]) + (1 - w) * m[1].predict(X[FEATS]), 0, None)
|
|
|
|
|
|
def next_month(hist, items, t_new=48, op_mult=1.0, ex=0):
|
|
"""다음 달(t_new) 예측용 행: 작년 같은 달 운용시간 x 배수, 훈련 여부 지정."""
|
|
last = hist[hist["t"] == t_new - 1].copy()
|
|
prev = hist[hist["t"] == t_new - 12].set_index("item_id")["op_hours"]
|
|
m = (t_new % 12) + 1
|
|
f = last.assign(t=t_new, ym=f"{2022 + t_new // 12}-{m:02d}", month=m, demand=0, exercise=ex,
|
|
op_hours=last["item_id"].map(prev).to_numpy() * op_mult)
|
|
X = make_features(pd.concat([hist, f], ignore_index=True))
|
|
return X[X["t"] == t_new].set_index("item_id").loc[items["item_id"]].reset_index()
|
|
|
|
|
|
def stock_policy(items, mu, sigma, z=1.65, review=1):
|
|
"""안전재고 = z x 월 예측오차 표준편차 x sqrt(리드타임+검토주기), 재주문점 = 월 수요 x (리드타임+검토주기) + 안전재고."""
|
|
L = items["lead_time"].to_numpy() + review
|
|
ss = z * sigma * np.sqrt(L)
|
|
return pd.DataFrame({"item_id": items["item_id"], "mu": mu, "sigma": sigma, "L": L,
|
|
"ss": np.ceil(ss), "rop": np.ceil(mu * L + ss)})
|
|
|
|
|
|
def risk_index(items, pol, stock):
|
|
"""결품 위험 지수(0~100): 리드타임 동안 수요가 가용 재고를 넘을 확률(정규 근사)을 기본으로,
|
|
단종 상태·단일 공급업체·대체품 부재에 가중을 준다(가상 가중치)."""
|
|
st = stock.set_index("item_id").loc[items["item_id"]]
|
|
avail = (st["on_hand"] + st["on_order"]).to_numpy()
|
|
p = 1 - norm.cdf((avail - pol["mu"].to_numpy() * pol["L"].to_numpy()) / (pol["sigma"].to_numpy() * np.sqrt(pol["L"].to_numpy()) + 1e-6))
|
|
sw = items["status"].map(STATUS_W).to_numpy()
|
|
sup = (items["suppliers"].to_numpy() == 1).astype(float)
|
|
alt = (~items["alt_available"].to_numpy()).astype(float)
|
|
score = 100 * np.clip(0.55 * p + 0.25 * sw + 0.10 * sup + 0.10 * alt, 0, 1)
|
|
return pd.DataFrame({"item_id": items["item_id"], "p_short": p, "status_w": sw, "single_sup": sup, "no_alt": alt,
|
|
"avail": avail, "risk": score.round(1)})
|
|
|
|
LTB_MONTHS = 24 # [교육용가상자료] 단종예고 품목 최종 구매(LTB) 대상 기간(개월)
|
|
t_all = time.perf_counter()
|
|
checks = []
|
|
|
|
# 1) 자료 수집·정합성
|
|
items = make_items()
|
|
hist = make_history(items)
|
|
stock = make_stock(items, hist)
|
|
ok = (hist["demand"] >= 0).all() and items["nsn"].str.match(r"^\d{4}-\d{2}-\d{3}-\d{4}$").all() and not hist.duplicated(["item_id", "t"]).any()
|
|
checks.append(("원천 자료 정합성(NSN 형식·음수·중복)", ok))
|
|
df = make_features(hist)
|
|
print(f"[1] 품목 {len(items)}개 x {N_MONTHS}개월 청구 이력, 학습 행 {len(df)}개, 정합성 {'통과' if ok else '실패'}")
|
|
|
|
# 2) 재학습 파이프라인: 운영 모델 v1(t<36) vs 재학습 v2(t<42), 최근 6개월(t=42~47) 검증 후 교체 판단
|
|
va = df[df["t"] >= 42]
|
|
big = va["item_id"].isin(items.loc[items["base"] >= 10, "item_id"]).to_numpy()
|
|
v1, v2 = fit_ens(df[df["t"] < 36]), fit_ens(df[df["t"] < 42])
|
|
p1, p2 = pred_ens(v1, va), pred_ens(v2, va)
|
|
e1, e2 = mape(va["demand"][big], p1[big]), mape(va["demand"][big], p2[big])
|
|
eb = mape(va["demand"][big], va["ma3"][big])
|
|
use = "v2" if e2 <= e1 else "v1"
|
|
print(f"[2] 검증 MAPE(수요 많은 품목): MA3 {eb:.1%}, 운영 v1 {e1:.1%}, 재학습 v2 {e2:.1%} -> {use} {'교체' if use == 'v2' else '유지'}")
|
|
checks += [("자동 재학습 실행·비교", True), ("운영 모델 MAPE 25% 이하", min(e1, e2) <= 0.25)]
|
|
|
|
# 3) 다음 달 예측(전체 자료로 최종 학습) -> 안전재고·위험 지수
|
|
final = fit_ens(df)
|
|
mu = pred_ens(final, next_month(hist, items))
|
|
sg = np.maximum(0.25 * mu, np.sqrt(np.maximum(mu, 1)))
|
|
pol = stock_policy(items, mu, sg)
|
|
rk = risk_index(items, pol, stock)
|
|
alert = rk["risk"] >= 60
|
|
print(f"[3] 다음 달 예측 수요 {mu.sum():.0f}개, 결품 위험 경보(60 이상) {int(alert.sum())}개 품목, 안전재고 합 {pol['ss'].sum():.0f}개")
|
|
|
|
# 4) 단종 대응: 단종예고 -> 최종 구매(LTB) 수량, 단종 -> 대체품·재생정비 검토
|
|
ip = (stock["on_hand"] + stock["on_order"]).to_numpy()
|
|
eol = items.assign(mu=mu, ip=ip, risk=rk["risk"].to_numpy())
|
|
ltb = eol[eol["status"] == "단종예고"].assign(ltb=lambda d: np.maximum(0, np.ceil(d["mu"] * LTB_MONTHS - d["ip"])))
|
|
dis = eol[eol["status"] == "단종"]
|
|
print(f"[4] 단종예고 {len(ltb)}개 품목 최종 구매 추천 합 {ltb['ltb'].sum():.0f}개(향후 {LTB_MONTHS}개월분), 단종 {len(dis)}개 품목 중 대체품 보유 {int(dis['alt_available'].sum())}개")
|
|
for _, r in ltb.sort_values("risk", ascending=False).head(3).iterrows():
|
|
print(f" {r['item_id']}({r['group']}) 월 {r['mu']:.1f}개, 가용 {r['ip']:.0f}개 -> 최종 구매 {r['ltb']:.0f}개")
|
|
|
|
# 5) 예산 제약 발주 추천(LP, 생산중 품목. 단종예고는 최종 구매로 별도 처리) -> 결재 대기열
|
|
gap = np.where(items["status"] == "생산중", np.maximum(0, np.ceil(pol["rop"].to_numpy() - ip)), 0)
|
|
price = items["price"].to_numpy().astype(float)
|
|
idx = np.where(gap > 0)[0]
|
|
budget = 0.6 * float((gap * price).sum())
|
|
lp = linprog(c=-(rk["risk"].to_numpy()[idx] + 10) / gap[idx], A_ub=[price[idx]], b_ub=[budget], bounds=[(0, g) for g in gap[idx]], method="highs")
|
|
q = np.zeros(len(items)); q[idx] = np.floor(lp.x)
|
|
queue = [{"id": f"Q{k + 1:02d}", "item": items.loc[i, "item_id"], "qty": int(q[i]), "risk": float(rk.loc[i, "risk"]), "status": "결재대기"} for k, i in enumerate(np.where(q > 0)[0])]
|
|
queue += [{"id": f"Q{len(queue) + k + 1:02d}", "item": r["item_id"], "qty": int(r["ltb"]), "risk": float(r["risk"]), "status": "결재대기", "type": "LTB"} for k, (_, r) in enumerate(ltb[ltb["ltb"] > 0].iterrows())]
|
|
auto = [x for x in queue if x["status"] != "결재대기"]
|
|
print(f"[5] 발주 추천 {int((q > 0).sum())}건(예산 {budget / 1e6:.1f}백만 중 {np.dot(q, price) / 1e6:.1f}백만) + 최종 구매 {int((ltb['ltb'] > 0).sum())}건 -> 정비장교 결재 대기 {len(queue)}건, 자동 확정 {len(auto)}건")
|
|
checks.append(("자동 확정 0건(HITL)", not auto))
|
|
|
|
# 6) 운영 가이드·연계 메시지
|
|
GUIDE = {"1. 목적·범위": "야전 정비 부품 수요 예측·재고 최적화·단종 대응", "2. 데이터 규격": f"품목 마스터 {items.shape[1]}개 필드, NSN 가상 형식",
|
|
"3. 전처리": "NSN 정규화, 결측 월 계절성 보간, 긴급 불출 보존", "4. 예측 모델": f"포아송 GBM+RF 앙상블, MAPE {min(e1, e2):.1%}",
|
|
"5. 안전재고·위험 지수": "z=1.65, 리드타임 결품 확률+단종·공급업체·대체품", "6. 단종 대응": f"LTB {LTB_MONTHS}개월분, 대체품 호환성 검증",
|
|
"7. 결재선·감사": "자동 확정 금지, 승인·조정·반려·보류 사유 기록", "8. 재학습": "분기별 재학습, 검증 MAPE가 낮을 때만 교체",
|
|
"9. 폭증 대응": "작전 템포 계수 보정, 경계 구간 '주의' 표시", "10. 연계 인터페이스": "ORDER_REQ 가상 메시지 규격"}
|
|
done = np.mean([bool(v) for v in GUIDE.values()])
|
|
checks.append(("운영 가이드 10개 장 완비", done == 1))
|
|
print("[6] 운영 가이드 목차:", " / ".join(GUIDE))
|
|
top = max(queue, key=lambda x: x["risk"]) if queue else {}
|
|
msg = {"msg_type": "ORDER_REQ", "version": "1.0", "item_id": top.get("item"), "nsn": items.set_index("item_id").loc[top.get("item"), "nsn"] if top else None,
|
|
"qty": top.get("qty"), "risk_index": top.get("risk"), "model_ver": use, "approval": "PENDING", "source": "교육용가상자료"}
|
|
print("[7] 체계 연계 메시지 예시(가상 규격):", json.dumps(msg, ensure_ascii=False))
|
|
elapsed = time.perf_counter() - t_all
|
|
checks.append(("통합 실행 60초 이내", elapsed < 60))
|
|
|
|
print("\n[자체 점검표]")
|
|
for name, okk in checks:
|
|
print(f" [{'PASS' if okk else 'FAIL'}] {name}")
|
|
print(f"점검 {sum(o for _, o in checks)}/{len(checks)} 통과, 운영 가이드 완비율 {done:.0%}, 전체 실행 {elapsed:.1f}s")
|
|
print("\n[브리핑 요약]")
|
|
print(f"- 부품 수요 예측 MAPE {min(e1, e2):.1%}(이동평균 {eb:.1%})이며, 재학습 모델은 검증 후 교체한다({use} 사용).")
|
|
print(f"- 결품 위험 경보 {int(alert.sum())}개 품목 중 단종·단종예고가 {int((alert & (items['status'] != '생산중')).sum())}개로, 단종 대응이 재고 위험의 핵심이다.")
|
|
print(f"- 단종예고 {len(ltb)}개 품목의 최종 구매 {ltb['ltb'].sum():.0f}개와 발주 {int((q > 0).sum())}건을 결재 대기열에 올렸다.")
|
|
print("- 모든 발주·최종 구매·대체품 전환은 정비장교 결재 후 집행하며, 수치는 교육용 가상 자료 기준이다.")
|
|
|
|
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
|
|
ax[0].bar(["MA3", "model v1", "model v2"], [eb, e1, e2], color=["gray", "tab:blue", "tab:green"]); ax[0].axhline(0.25, c="r", ls="--")
|
|
ax[0].set_title("Validation MAPE (t=42-47)")
|
|
col = items["status"].map({"생산중": "tab:green", "단종예고": "tab:orange", "단종": "tab:red"})
|
|
ax[1].scatter(items["lead_time"], rk["risk"], c=col, s=20 + mu / 2); ax[1].axhline(60, c="r", ls="--")
|
|
ax[1].set_xlabel("lead time (months)"); ax[1].set_ylabel("risk index"); ax[1].set_title("Risk vs lead time (size=demand, red=discontinued)")
|
|
plt.tight_layout(); plt.show()
|
|
```
|
|
|
|
실행 결과 예시
|
|
|
|
```text
|
|
[1] 품목 40개 x 48개월 청구 이력, 학습 행 1000개, 정합성 통과
|
|
[2] 검증 MAPE(수요 많은 품목): MA3 32.7%, 운영 v1 20.0%, 재학습 v2 20.9% -> v1 유지
|
|
[3] 다음 달 예측 수요 948개, 결품 위험 경보(60 이상) 13개 품목, 안전재고 합 1108개
|
|
[4] 단종예고 8개 품목 최종 구매 추천 합 2779개(향후 24개월분), 단종 3개 품목 중 대체품 보유 2개
|
|
P004(filter) 월 8.1개, 가용 12개 -> 최종 구매 183개
|
|
P019(filter) 월 11.5개, 가용 59개 -> 최종 구매 219개
|
|
P025(filter) 월 14.0개, 가용 89개 -> 최종 구매 247개
|
|
[5] 발주 추천 10건(예산 31.2백만 중 30.9백만) + 최종 구매 8건 -> 정비장교 결재 대기 18건, 자동 확정 0건
|
|
[6] 운영 가이드 목차: 1. 목적·범위 / 2. 데이터 규격 / 3. 전처리 / 4. 예측 모델 / 5. 안전재고·위험 지수 / 6. 단종 대응 / 7. 결재선·감사 / 8. 재학습 / 9. 폭증 대응 / 10. 연계 인터페이스
|
|
[7] 체계 연계 메시지 예시(가상 규격): {"msg_type": "ORDER_REQ", "version": "1.0", "item_id": "P004", "nsn": "2940-00-559-8893", "qty": 183, "risk_index": 87.5, "model_ver": "v1", "approval": "PENDING", "source": "교육용가상자료"}
|
|
|
|
[자체 점검표]
|
|
[PASS] 원천 자료 정합성(NSN 형식·음수·중복)
|
|
[PASS] 자동 재학습 실행·비교
|
|
[PASS] 운영 모델 MAPE 25% 이하
|
|
[PASS] 자동 확정 0건(HITL)
|
|
[PASS] 운영 가이드 10개 장 완비
|
|
[PASS] 통합 실행 60초 이내
|
|
점검 6/6 통과, 운영 가이드 완비율 100%, 전체 실행 3.5s
|
|
|
|
[브리핑 요약]
|
|
- 부품 수요 예측 MAPE 20.0%(이동평균 32.7%)이며, 재학습 모델은 검증 후 교체한다(v1 사용).
|
|
- 결품 위험 경보 13개 품목 중 단종·단종예고가 11개로, 단종 대응이 재고 위험의 핵심이다.
|
|
- 단종예고 8개 품목의 최종 구매 2779개와 발주 10건을 결재 대기열에 올렸다.
|
|
- 모든 발주·최종 구매·대체품 전환은 정비장교 결재 후 집행하며, 수치는 교육용 가상 자료 기준이다.
|
|
```
|