miliai/pbl02
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
pbl02/scenario_md/MOV01-수송·물류·창고_수송수요-데이터 기반 수송 소요예측 Tool/MOV01_정답코드.md
T
KIM SU BO 350bb92bf9 시나리오 내용 반영
2026-10-08 16:43:48 +09:00

2766 lines
167 KiB
Markdown
Raw Permalink Blame History

# MOV01 정답 코드 - 데이터 기반 수송 소요예측 Tool
> [교육용가상자료] 모든 데이터는 코드 안에서 난수로 만든 가상 자료이며 실제 부대·장비·인원·작전 정보와 무관합니다.
## 실행 환경
- Python 3.10 이상, numpy, pandas, scipy, scikit-learn, matplotlib
- 미션별 코드는 각각 한 파일로 독립 실행됩니다. 외부 파일 없이 코드 안에서 가상 데이터를 생성합니다.
- 그래프의 글자는 한글 글꼴이 없는 실행 환경에서도 깨지지 않도록 영문으로 표기했습니다. 표·출력 결과는 한글입니다.
- 난수 시드를 고정해 두어 같은 코드를 실행하면 같은 결과가 나옵니다.
## 미션 1. 전시 주요 보급로(MSR) 교차로, 병목 구간(병목 다리/터널) 및 통과 능력 데이터 정의
- 난이도: 초급 / 예상 소요시간: 1.5시간
- 코드 설명: 가상 노선망의 교차로·병목(교량·터널) 구간과 통과 능력 데이터를 정의하고, 수송 소요예측 업무 프로세스 맵과 20개 이상의 피처 데이터 사전을 만들어 실제 데이터 컬럼과 대조한다.
```python
"""가상 노선망의 교차로·병목(교량·터널) 구간과 통과 능력 데이터를 정의하고, 수송 소요예측 업무 프로세스 맵과 20개 이상의 피처 데이터 사전을 만들어 실제 데이터 컬럼과 대조한다."""
import re
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# 1) 데이터 사전: 원천 컬럼 + 파생 피처 (이름, 구분, 자료형, 허용 범위/값, 설명)
DICT = [
("seg", "원천", "str", r"^[ABCX]\d$", "구간 코드(가상 노선망)"), ("kind", "원천", "str", "일반|교량|터널|교차로", "구간 유형(병목 후보)"),
("km", "원천", "float", "1~50", "구간 연장(km)"), ("cap", "원천", "float", "10~300", "피크시간 통과 용량(대/시)"),
("t0", "원천", "float", "1~120", "자유 주행 시간(분)"), ("day", "원천", "int", "0~", "관측 일자 번호"),
("dow", "원천", "int", "0~6", "요일 번호"), ("plan_v", "원천", "float", "0~", "수송계획 피크 물량(대/시)"),
("v", "원천", "float", "0~", "실제 피크 통과량(대/시)"), ("maint", "원천", "int", "0|1", "비계획 정비 여부(통과 용량 감소)"),
("urgent", "원천", "int", "0|1", "긴급 불출 여부(물량 증가)"), ("rain", "원천", "int", "0|1", "우천 여부"),
("cap_eff", "원천", "float", "0~cap", "정비·기상 반영 유효 용량"), ("delay", "목표", "float", "0~", "구간 지체 시간(분)"),
("transit", "목표", "float", "t0~", "구간 통과시간(분)"),
("util", "파생", "float", "0~2", "계획물량/공칭용량"), ("util_eff", "파생", "float", "0~3", "계획물량/유효용량"),
("util_eff4", "파생", "float", "0~", "유효 이용률 4제곱(혼잡 비선형)"), ("bpr_base", "파생", "float", "0~", "BPR 기준 지체(분)"),
("lag1_delay", "파생", "float", "0~", "전일 지체"), ("lag7_delay", "파생", "float", "0~", "전주 동요일 지체"),
("roll7_v", "파생", "float", "0~", "최근 7일 평균 통과량"), ("surge_ratio", "파생", "float", "0~", "계획물량/최근 평균(폭증 지표)"),
("is_weekend", "파생", "int", "0|1", "주말 여부"), ("dow_sin", "파생", "float", "-1~1", "요일 순환(사인)"),
("dow_cos", "파생", "float", "-1~1", "요일 순환(코사인)"), ("k_교량", "파생", "int", "0|1", "교량 구간 더미"),
("k_터널", "파생", "int", "0|1", "터널 구간 더미"), ("k_교차로", "파생", "int", "0|1", "교차로 구간 더미")]
dd = pd.DataFrame(DICT, columns=["필드", "구분", "형", "허용값", "설명"])
print(f"[1] 데이터 사전 {len(dd)}개 필드 (원천 {int((dd.구분 == '원천').sum())}, 목표 {int((dd.구분 == '목표').sum())}, 파생 {int((dd.구분 == '파생').sum())})")
# 2) 실제 생성 데이터와 대조: 사전에 없는 컬럼 / 사전에만 있는 필드
df = make_data()
feat = add_features(df)
miss = sorted(set(feat.columns) - set(dd["필드"]))
extra = sorted(set(dd["필드"]) - set(feat.columns))
print(f"[2] 데이터 사전 대조: 사전 누락 컬럼 {miss or '없음'}, 데이터에 없는 필드 {extra or '없음'}")
fe = dd[dd["구분"] != "목표"]
print(f" 예측 피처 후보(원천+파생 중 목표 제외) {len(fe)}개 -> 목표(20개 이상) {'달성' if len(fe) >= 20 else '미달'}")
bad_rng = {"v": int((feat.v < 0).sum()), "cap_eff": int((feat.cap_eff > feat.cap).sum()), "util_eff": int((feat.util_eff > 3).sum())}
print(" 허용 범위 위반 건수:", bad_rng)
# 3) 구간 마스터: 병목 후보(교량·터널·교차로)와 통과 능력
seg = SEGS.copy()
seg["병목후보"] = np.where(seg["kind"].isin(["교량", "터널", "교차로"]), "Y", "")
seg["평시 이용률"] = seg["seg"].map(UTIL)
print("\n[3] 구간 마스터(가상 노선망, 교육용 가정값)")
print(seg[["seg", "u", "v", "km", "kind", "cap", "t0", "평시 이용률", "병목후보"]].round(2).to_string(index=False))
# 4) 수송 소요예측 업무 프로세스 맵(단계, 담당, 입력 데이터, 산출, 현행 소요 분)
proc = [("수송 소요 접수", "수송계획장교", "부대별 요청 서식", "일일 수송 소요표", 30), ("구간 통과량 집계", "수송통제소", "구간 통과 보고", "plan_v, v", 40),
("정비·기상 정보 취합", "통제소/기상반", "정비 공지, 예보", "maint, rain", 25), ("통과 용량 산정", "수송통제소", "구간 마스터", "cap, cap_eff", 20),
("지체 예측", "분석 담당", "피처 테이블", "delay 예측", 45), ("병목 판정", "분석 담당", "예측 지체, 이용률", "병목 목록", 20),
("우회 검토·승인", "작전군수관", "우회 후보, 추가 소요", "우회 명령(승인)", 30), ("명령 하달·운행", "수송통제소", "승인된 명령", "운행 이벤트", 15),
("실적 대조·보정", "분석 담당", "예측 대 실적", "오차 보고", 35)]
pm = pd.DataFrame(proc, columns=["단계", "담당", "입력", "산출", "현행(분)"])
pm.index = range(1, len(pm) + 1)
print(f"\n[4] 업무 프로세스 맵 {len(pm)}단계, 현행 합계 {pm['현행(분)'].sum()}분 (가장 긴 단계: {pm.loc[pm['현행(분)'].idxmax(), '단계']})")
print(pm.to_string())
print("\n프로세스 맵 작성 완료(9단계), 피처 목록 20개 이상 -> 목표", "달성" if len(pm) >= 5 and len(fe) >= 20 else "미달")
print("모든 노선·구간·수치는 교육용 가상 자료이며 실제 보급로·지명·부대와 무관하다.")
```
실행 결과 예시
```text
[1] 데이터 사전 29개 필드 (원천 13, 목표 2, 파생 14)
[2] 데이터 사전 대조: 사전 누락 컬럼 없음, 데이터에 없는 필드 없음
예측 피처 후보(원천+파생 중 목표 제외) 27개 -> 목표(20개 이상) 달성
허용 범위 위반 건수: {'v': 0, 'cap_eff': 0, 'util_eff': 0}
[3] 구간 마스터(가상 노선망, 교육용 가정값)
seg u v km kind cap t0 평시 이용률 병목후보
A1 SRC J1 12 일반 130 13.09 0.78
A2 J1 J2 8 교량 70 12.00 0.85 Y
A3 J2 DST 10 교차로 140 13.33 0.66 Y
B1 SRC J3 15 일반 100 16.36 0.70
B2 J3 J2 9 터널 80 12.00 0.72 Y
B3 J3 DST 14 일반 95 15.27 0.64
C1 SRC J4 18 일반 70 19.64 0.55
C2 J4 DST 20 교량 75 30.00 0.58 Y
X1 J1 J3 7 교차로 50 9.33 0.45 Y
X2 J4 J3 10 일반 60 10.91 0.40
[4] 업무 프로세스 맵 9단계, 현행 합계 260분 (가장 긴 단계: 지체 예측)
단계 담당 입력 산출 현행(분)
1 수송 소요 접수 수송계획장교 부대별 요청 서식 일일 수송 소요표 30
2 구간 통과량 집계 수송통제소 구간 통과 보고 plan_v, v 40
3 정비·기상 정보 취합 통제소/기상반 정비 공지, 예보 maint, rain 25
4 통과 용량 산정 수송통제소 구간 마스터 cap, cap_eff 20
5 지체 예측 분석 담당 피처 테이블 delay 예측 45
6 병목 판정 분석 담당 예측 지체, 이용률 병목 목록 20
7 우회 검토·승인 작전군수관 우회 후보, 추가 소요 우회 명령(승인) 30
8 명령 하달·운행 수송통제소 승인된 명령 운행 이벤트 15
9 실적 대조·보정 분석 담당 예측 대 실적 오차 보고 35
프로세스 맵 작성 완료(9단계), 피처 목록 20개 이상 -> 목표 달성
모든 노선·구간·수치는 교육용 가상 자료이며 실제 보급로·지명·부대와 무관하다.
```
## 미션 2. 수송 수량 폭증에 따른 구간별 병목 지체 시간(Delay) 계산 전처리 모듈 구현
- 난이도: 초급~중급 / 예상 소요시간: 2.0시간
- 코드 설명: 수송량 폭증에 따른 구간별 지체 시간 계산 전처리 모듈: 오염된 구간 통과 기록에 정제 규칙 10종을 적용하고 비계획 정비·긴급 불출 같은 군수 특수 예외는 삭제하지 않고 보존하며, 정합성 검증과 전후 통계를 비교한다.
```python
"""수송량 폭증에 따른 구간별 지체 시간 계산 전처리 모듈: 오염된 구간 통과 기록에 정제 규칙 10종을 적용하고 비계획 정비·긴급 불출 같은 군수 특수 예외는 삭제하지 않고 보존하며, 정합성 검증과 전후 통계를 비교한다."""
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
clean = make_data()
rng = np.random.default_rng(2)
dirty = clean.copy()
# ---- 오염 주입(가상): 결측·음수·단위 오류·표기 오류·미등록 구간·0 용량·중복 ----
ix = rng.choice(len(dirty), 40, replace=False)
dirty.loc[ix[:12], "v"] = np.nan
dirty.loc[ix[12:18], "v"] = -dirty.loc[ix[12:18], "v"]
dirty.loc[ix[18:23], "v"] = dirty.loc[ix[18:23], "v"] * 10
dirty.loc[ix[23:29], "transit"] = np.nan
dirty.loc[ix[29:32], "cap"] = 0
dirty.loc[ix[32:36], "seg"] = dirty.loc[ix[32:36], "seg"].str.lower() + " "
ghost = dirty.iloc[ix[36:40]].copy(); ghost["seg"] = "Z9" # 마스터에 없는 구간 코드가 섞여 들어온 행
dirty = pd.concat([dirty, ghost, dirty.iloc[rng.choice(len(dirty), 15, replace=False)]], ignore_index=True)
planted = {"결측 통과량": 12, "음수 통과량": 6, "단위 오류(10배)": 5, "결측 통과시간": 6, "용량 0": 3, "구간 표기 오류": 4, "미등록 구간": 4, "중복 행": 15}
print(f"[0] 오염 데이터 {len(dirty)}행 (원본 {len(clean)}행), 주입 오류 {sum(planted.values())}건: {planted}")
log, w = [], dirty.copy()
master = SEGS.set_index("seg")
def rule(name, n):
log.append((len(log) + 1, name, n))
# R1 구간 코드 표기 정규화
before = w["seg"].copy(); w["seg"] = w["seg"].str.strip().str.upper(); rule("구간 코드 공백·대소문자 정규화", int((before != w["seg"]).sum()))
# R2 미등록 구간 격리(삭제하지 않고 별도 보관)
bad = ~w["seg"].isin(master.index); quarantine = w[bad].copy(); w = w[~bad]; rule("미등록 구간 격리(별도 보관)", int(bad.sum()))
# R3 중복 제거(구간·일자 키)
n0 = len(w); w = w.drop_duplicates(["seg", "day"], keep="first"); rule("(구간, 일자) 중복 제거", n0 - len(w))
# R4 용량 0/결측은 구간 마스터에서 복원
bad = w["cap"] <= 0; w.loc[bad, "cap"] = w.loc[bad, "seg"].map(master["cap"]); rule("용량 0 -> 구간 마스터 복원", int(bad.sum()))
# R5 음수 통과량은 결측으로 전환(부호 오류 가능성 -> 보간 대상)
bad = w["v"] < 0; w.loc[bad, "v"] = np.nan; rule("음수 통과량 -> 결측 처리", int(bad.sum()))
# R6 단위 오류(용량의 3배 초과)는 1/10 보정
bad = w["v"] > 3 * w["cap"]; w.loc[bad, "v"] = w.loc[bad, "v"] / 10; rule("단위 오류(용량 3배 초과) 1/10 보정", int(bad.sum()))
# R7 결측 통과량: 같은 구간·같은 요일의 중앙값으로 보간(계절성 보간)
bad = w["v"].isna(); med = w.groupby(["seg", "dow"])["v"].transform("median"); w.loc[bad, "v"] = med[bad]; rule("결측 통과량 -> 구간·요일 중앙값 보간", int(bad.sum()))
# R8 결측 통과시간: 자유주행 + 지체로 재계산
bad = w["transit"].isna(); w.loc[bad, "transit"] = w.loc[bad, "t0"] * (1 + 0.08 * w.loc[bad, "rain"]) + w.loc[bad, "delay"]; rule("결측 통과시간 재계산", int(bad.sum()))
# R9 지체 시간 정합: delay = transit - 자유주행(우천 보정), 음수는 0으로
d2 = (w["transit"] - w["t0"] * (1 + 0.08 * w["rain"])).clip(lower=0); mism = (d2 - w["delay"]).abs() > 1e-6
w["delay"] = d2; rule("지체 = 통과시간 - 자유주행 정합 재계산", int(mism.sum()))
# R10 이상치 표시: IQR 밖 지체는 '표시'만 하고, 정비·긴급 불출 행은 예외로 보존(삭제 금지)
q1, q3 = w["delay"].quantile([.25, .75]); out = w["delay"] > q3 + 3 * (q3 - q1)
w["outlier"] = out.astype(int); w["exception_keep"] = ((w["maint"] == 1) | (w["urgent"] == 1)).astype(int)
rule("지체 이상치 플래그(삭제 없음)", int(out.sum()))
w = w.sort_values(["day", "seg"]).reset_index(drop=True)
print("\n[1] 정제 규칙 적용 로그")
print(pd.DataFrame(log, columns=["규칙", "내용", "처리 건수"]).to_string(index=False))
print(f" 정제 규칙 {len(log)}종 적용(목표 10종: {'달성' if len(log) >= 10 else '미달'})")
# ---- 정합성 검증 ----
checks = {"(구간, 일자) 유일": not w.duplicated(["seg", "day"]).any(), "필수 컬럼 결측 0": int(w[["seg", "v", "cap", "delay", "transit"]].isna().sum().sum()) == 0,
"구간 코드 마스터 포함": bool(w["seg"].isin(master.index).all()), "통과량 0 이상·용량 3배 이하": bool(((w["v"] >= 0) & (w["v"] <= 3 * w["cap"])).all()),
"용량 = 마스터 값": bool((w["cap"] == w["seg"].map(master["cap"])).all()), "통과시간 >= 자유주행": bool((w["transit"] >= w["t0"] - 1e-9).all()),
"지체 >= 0": bool((w["delay"] >= 0).all())}
passed = sum(checks.values())
print("\n[2] 정합성 검증")
for k, v in checks.items():
print(f" [{'PASS' if v else 'FAIL'}] {k}")
print(f" 통과율 {passed / len(checks):.0%} (목표 100%)")
# ---- 군수 특수 예외 보존 / 원본 대비 복원 정확도 ----
ex = w[w["exception_keep"] == 1]
print(f"\n[3] 군수 특수 예외 보존: 비계획 정비 {int(w['maint'].sum())}건, 긴급 불출 {int(w['urgent'].sum())}건 모두 유지(원본 {int(clean['maint'].sum())}, {int(clean['urgent'].sum())}건)")
print(f" 예외 행 평균 지체 {ex['delay'].mean():.2f}분 vs 일반 행 {w.loc[w['exception_keep'] == 0, 'delay'].mean():.2f}분 -> 이상치로 지웠다면 폭증 구간 정보가 사라진다")
m = w.merge(clean[["seg", "day", "v"]], on=["seg", "day"], suffixes=("", "_true"))
print(f" 통과량 복원 오차(보정·보간된 행 {int((m['v'] != m['v_true']).sum())}건 포함 전체 MAE) {np.abs(m['v'] - m['v_true']).mean():.3f}대/시")
print("\n[4] 전후 기술통계 비교(통과량 v / 지체 delay)")
st = pd.DataFrame({"오염 원본": dirty[["v", "delay"]].describe().loc[["count", "mean", "min", "max"]].stack(), "정제 후": w[["v", "delay"]].describe().loc[["count", "mean", "min", "max"]].stack(),
"정답 원본": clean[["v", "delay"]].describe().loc[["count", "mean", "min", "max"]].stack()}).round(2)
print(st.to_string())
```
실행 결과 예시
```text
[0] 오염 데이터 1519행 (원본 1500행), 주입 오류 55건: {'결측 통과량': 12, '음수 통과량': 6, '단위 오류(10배)': 5, '결측 통과시간': 6, '용량 0': 3, '구간 표기 오류': 4, '미등록 구간': 4, '중복 행': 15}
[1] 정제 규칙 적용 로그
규칙 내용 처리 건수
1 구간 코드 공백·대소문자 정규화 4
2 미등록 구간 격리(별도 보관) 4
3 (구간, 일자) 중복 제거 15
4 용량 0 -> 구간 마스터 복원 3
5 음수 통과량 -> 결측 처리 6
6 단위 오류(용량 3배 초과) 1/10 보정 5
7 결측 통과량 -> 구간·요일 중앙값 보간 18
8 결측 통과시간 재계산 6
9 지체 = 통과시간 - 자유주행 정합 재계산 0
10 지체 이상치 플래그(삭제 없음) 43
정제 규칙 10종 적용(목표 10종: 달성)
[2] 정합성 검증
[PASS] (구간, 일자) 유일
[PASS] 필수 컬럼 결측 0
[PASS] 구간 코드 마스터 포함
[PASS] 통과량 0 이상·용량 3배 이하
[PASS] 용량 = 마스터 값
[PASS] 통과시간 >= 자유주행
[PASS] 지체 >= 0
통과율 100% (목표 100%)
[3] 군수 특수 예외 보존: 비계획 정비 95건, 긴급 불출 87건 모두 유지(원본 95, 87건)
예외 행 평균 지체 4.37분 vs 일반 행 1.60분 -> 이상치로 지웠다면 폭증 구간 정보가 사라진다
통과량 복원 오차(보정·보간된 행 18건 포함 전체 MAE) 0.104대/시
[4] 전후 기술통계 비교(통과량 v / 지체 delay)
오염 원본 정제 후 정답 원본
count v 1507.00 1500.00 1500.00
delay 1519.00 1500.00 1500.00
mean v 63.64 62.95 62.95
delay 1.94 1.93 1.93
min v -146.18 16.78 16.78
delay 0.00 0.00 0.00
max v 635.70 174.46 174.46
delay 21.79 21.79 21.79
```
## 미션 3. 구간별 통과 용량 기반 1차 수송 지체 예측 Baseline 알고리즘 구현
- 난이도: 중급 / 예상 소요시간: 2.5시간
- 코드 설명: 구간별 통과 용량(BPR 함수) 기반 1차 수송 지체 예측 Baseline을 만들어 오차를 측정하고, EDA 차트 8종과 상관 피처·파레토(상위 20% 구간) 분석으로 Baseline의 한계를 찾는다.
```python
"""구간별 통과 용량(BPR 함수) 기반 1차 수송 지체 예측 Baseline을 만들어 오차를 측정하고, EDA 차트 8종과 상관 피처·파레토(상위 20% 구간) 분석으로 Baseline의 한계를 찾는다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import stats
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_data()
d = add_features(df)
# 1) Baseline: 계획 물량 / 공칭 용량으로 BPR 지체 계산(정비·기상·전일 지체 미반영)
d["pred_base"] = d["bpr_base"]
d["pred_transit"] = d["t0"] + d["pred_base"]
print(f"[1] Baseline(BPR) 지체 예측: 통과시간 MAPE {mape(d.transit, d.pred_transit):.1f}%, 통과시간 MAE {mae(d.transit, d.pred_transit):.2f}분, 지체 MAE {mae(d.delay, d.pred_base):.2f}분")
bias = (d["pred_base"] - d["delay"])
grp = pd.DataFrame({"평균 오차(예측-실제, 분)": [bias[d.maint == 1].mean(), bias[d.maint == 0].mean(), bias[d.rain == 1].mean(), bias[d.rain == 0].mean(), bias[d.urgent == 1].mean(), bias[(d.day >= 100) & (d.day < 110)].mean()]},
index=["비계획 정비일", "정비 없음", "우천일", "맑음", "긴급 불출일", "수송량 폭증 기간"]).round(2)
print("[2] 조건별 Baseline 오차(음수 = 과소예측)")
print(grp.to_string())
print(" -> Baseline은 정비·긴급 불출·폭증 조건에서 지체를 과소예측한다(한계).")
# 2) 상관 분석: 지체와 각 피처의 스피어만 상관·p-value
rows = []
for f in FEATS:
r, p = stats.spearmanr(d[f], d["delay"])
rows.append((f, r, p))
cor = pd.DataFrame(rows, columns=["피처", "상관", "p"]).assign(abs_r=lambda x: x["상관"].abs()).sort_values("abs_r", ascending=False)
sig = cor[(cor["abs_r"] >= 0.3) & (cor["p"] < 0.05)]
print(f"\n[3] 지체와 유의한 상관(|r|>=0.3, p<0.05) 피처 {len(sig)}개 (목표 5개 이상: {'달성' if len(sig) >= 5 else '미달'})")
print(sig.head(7)[["피처", "상관", "p"]].round(3).to_string(index=False))
t, p = stats.ttest_ind(d.loc[d.maint == 1, "delay"], d.loc[d.maint == 0, "delay"], equal_var=False)
print(f" 가설검정: 비계획 정비일 vs 평시 지체 평균차 t={t:.1f}, p={p:.2g}")
# 3) 파레토(ABC): 구간별 총 지체 기여
tot = d.groupby("seg")["delay"].sum().sort_values(ascending=False)
cum = tot.cumsum() / tot.sum()
top = tot.index[:max(1, int(np.ceil(len(tot) * 0.2)))]
print(f"\n[4] 파레토: 상위 20% 구간({', '.join(top)})이 총 지체의 {tot[top].sum() / tot.sum():.0%}를 차지, 상위 50% 구간 {cum.iloc[4]:.0%}")
# 4) EDA 차트 8종
fig, ax = plt.subplots(2, 4, figsize=(18, 8))
daily = d.groupby("day")["v"].sum(); ax[0, 0].plot(daily.index, daily.values); ax[0, 0].set_title("1 Daily total peak volume")
ax[0, 1].hist(d["delay"], bins=40); ax[0, 1].set_title("2 Delay distribution")
ax[0, 2].scatter(d["util"], d["delay"], s=6, alpha=.4); ax[0, 2].set_title("3 Utilization vs delay")
mean_seg = d.groupby("seg")["delay"].mean().sort_values(); ax[0, 3].barh(mean_seg.index, mean_seg.values); ax[0, 3].set_title("4 Mean delay by segment")
ax[1, 0].boxplot([d.loc[d.dow == k, "delay"] for k in range(7)]); ax[1, 0].set_title("5 Delay by weekday")
ax[1, 1].boxplot([d.loc[d.rain == 0, "delay"], d.loc[d.rain == 1, "delay"]])
ax[1, 1].set_xticklabels(["dry", "rain"]); ax[1, 1].set_title("6 Rain effect")
ax[1, 2].boxplot([d.loc[d.maint == 0, "delay"], d.loc[d.maint == 1, "delay"]])
ax[1, 2].set_xticklabels(["normal", "maintenance"]); ax[1, 2].set_title("7 Maintenance effect")
ax[1, 3].bar(range(len(tot)), (tot / tot.sum()).values); ax[1, 3].plot(range(len(tot)), cum.values, "r-o"); ax[1, 3].set_title("8 Pareto of total delay")
plt.suptitle("MOV01 EDA (synthetic data)"); plt.tight_layout(); plt.show()
print("\n[5] EDA 차트 8종 생성 완료(목표 8종 이상: 달성)")
print("핵심 발견: 지체는 이용률의 비선형 함수이며 정비·폭증·긴급 불출이 겹치면 급증한다. 이는 M4 파생 피처 설계의 근거가 된다.")
```
실행 결과 예시
```text
[1] Baseline(BPR) 지체 예측: 통과시간 MAPE 7.4%, 통과시간 MAE 1.39분, 지체 MAE 1.23분
[2] 조건별 Baseline 오차(음수 = 과소예측)
평균 오차(예측-실제, 분)
비계획 정비일 -4.87
정비 없음 -0.96
우천일 -1.67
맑음 -1.14
긴급 불출일 -2.62
수송량 폭증 기간 -2.66
-> Baseline은 정비·긴급 불출·폭증 조건에서 지체를 과소예측한다(한계).
[3] 지체와 유의한 상관(|r|>=0.3, p<0.05) 피처 14개 (목표 5개 이상: 달성)
피처 상관 p
util_eff 0.836 0.0
util_eff4 0.836 0.0
bpr_base 0.830 0.0
util 0.793 0.0
lag1_delay 0.730 0.0
plan_v 0.653 0.0
lag7_delay 0.650 0.0
가설검정: 비계획 정비일 vs 평시 지체 평균차 t=8.2, p=1.7e-12
[4] 파레토: 상위 20% 구간(A2, A1)이 총 지체의 33%를 차지, 상위 50% 구간 70%
[5] EDA 차트 8종 생성 완료(목표 8종 이상: 달성)
핵심 발견: 지체는 이용률의 비선형 함수이며 정비·폭증·긴급 불출이 겹치면 급증한다. 이는 M4 파생 피처 설계의 근거가 된다.
```
## 미션 4. 네트워크 흐름(Max-Flow) 알고리즘 기반 병목 발생 지점 사전 식별 및 우회로 탐색 모델 고도화
- 난이도: 중급 / 예상 소요시간: 3.0시간
- 코드 설명: 최대유량(Max-Flow) 알고리즘으로 가상 노선망의 최소 절단(병목) 구간을 사전 식별하고 구간 폐쇄 시 우회 경로를 찾으며, 파생 피처 14종과 시계열 교차검증으로 지체 예측 기준 모델(MAPE)을 만든다.
```python
"""최대유량(Max-Flow) 알고리즘으로 가상 노선망의 최소 절단(병목) 구간을 사전 식별하고 구간 폐쇄 시 우회 경로를 찾으며, 파생 피처 14종과 시계열 교차검증으로 지체 예측 기준 모델(MAPE)을 만든다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import linprog
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import TimeSeriesSplit
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
# 1) 최대유량과 병목(최소 절단)
f0, cut0 = max_flow()
print(f"[1] 노선망 최대 통과능력 {f0:.0f}대/시, 최소 절단(병목) 구간: {', '.join(cut0)}")
gain = []
for s in CAP:
g, _ = max_flow(capmul={s: 1.2})
gain.append((s, SEGS.set_index("seg").loc[s, "kind"], CAP[s], g - f0))
gain = pd.DataFrame(gain, columns=["구간", "유형", "용량", "용량 20% 증설 시 증가(대/시)"]).sort_values("용량 20% 증설 시 증가(대/시)", ascending=False)
print("[2] 구간별 증설 민감도(병목일수록 증설 효과가 큼)")
print(gain.head(5).to_string(index=False))
# 2) 사전 병목 식별: 계획 물량 기준 이용률 >= 0.9인 구간 + 최소 절단 구간
df = make_data(); d = add_features(df)
today = d[d.day == d.day.max()].set_index("seg")
risk = today[(today.util_eff >= 0.9)].index.tolist()
print(f"\n[3] 최종일(day {int(d.day.max())}) 사전 병목 경보: 유효 이용률 0.9 이상 구간 {risk or '없음'}")
# 3) 구간 폐쇄 시 우회: 최대유량 변화와 우회 경로
D = 200
print(f"\n[4] 구간 폐쇄 시나리오(수요 {D}대/시): 우회 경로와 평균 통과시간")
base = assign(D)
rows = [("평시", "-", f0, base["avg_min"], "A·B·C 분산", "수용")]
for c in (("A2",), ("B2",), ("A3",)):
mf, cut = max_flow(c)
r = assign(D, closed=c)
top = max(r["paths"], key=r["paths"].get)
rows.append((f"{c[0]} 폐쇄", ",".join(cut), mf, r["avg_min"], "->".join(top), "수용" if D <= mf else "과부하"))
sc = pd.DataFrame(rows, columns=["시나리오", "새 병목", "최대유량", "평균 통과(분)", "주 경로", "수요 대비"]).round(1)
print(sc.to_string(index=False))
print(f" A2 폐쇄 시 평균 통과시간 +{rows[1][3] - rows[0][3]:.1f}분, 수요 {D}대/시 대비 최대유량 {rows[1][2]:.0f}대/시 -> {rows[1][5]}(과부하면 우회만으로 부족하므로 물량 조정·타 수단 검토가 필요)")
# 4) 파생 피처로 지체 예측 기준 모델: 시계열 교차검증(TimeSeriesSplit)
print(f"\n[5] 파생 피처 {len([f for f in FEATS if f not in ['plan_v','cap','cap_eff','t0','km','maint','rain']])}개 생성(lag·이용률·BPR·폭증비·요일 순환·구간 더미 등)")
days = np.sort(d.day.unique())
tscv = TimeSeriesSplit(n_splits=4)
res = []
for k, (a, b) in enumerate(tscv.split(days)):
tr, te = d[d.day.isin(days[a])], d[d.day.isin(days[b])]
m = GradientBoostingRegressor(n_estimators=150, max_depth=3, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["delay"])
p = te["t0"] * (1 + 0.08 * te["rain"]) + m.predict(te[FEATS])
pb = te["t0"] + te["bpr_base"]
res.append((k + 1, len(tr), len(te), mape(te["transit"], pb), mape(te["transit"], p)))
cv = pd.DataFrame(res, columns=["fold", "학습", "검증", "BPR MAPE(%)", "기준모델 MAPE(%)"]).round(2)
print(cv.to_string(index=False))
mm = cv["기준모델 MAPE(%)"].mean()
print(f" 교차검증 평균 MAPE {mm:.2f}% (목표 25% 이하: {'달성' if mm <= 25 else '미달'}), 폴드가 뒤로 갈수록 학습 구간이 길어져 미래 정보 누수가 없다")
m = GradientBoostingRegressor(n_estimators=150, max_depth=3, learning_rate=0.08, random_state=0).fit(d[FEATS], d["delay"])
imp = pd.Series(m.feature_importances_, index=FEATS).sort_values(ascending=False)
print(" 피처 중요도 상위 5:", ", ".join(f"{k} {v:.2f}" for k, v in imp.head(5).items()), f"/ 중요도 1% 미만 {int((imp < 0.01).sum())}개(차원 축소 후보)")
fig, ax = plt.subplots(1, 2, figsize=(12, 4.5))
for s, (u, v) in ARCS.items():
pos = {"SRC": (0, 1.5), "J1": (1, 3), "J2": (2, 3), "J3": (1.3, 1.5), "J4": (1, 0), "DST": (3, 1.5)}
ax[0].plot([pos[u][0], pos[v][0]], [pos[u][1], pos[v][1]], "r-" if s in cut0 else "gray", lw=2 + CAP[s] / 40)
ax[0].text((pos[u][0] + pos[v][0]) / 2, (pos[u][1] + pos[v][1]) / 2, s, fontsize=9)
ax[0].set_title("Network (red = min-cut bottleneck)"); ax[0].axis("off")
imp.head(10)[::-1].plot.barh(ax=ax[1]); ax[1].set_title("Feature importance (top 10)")
plt.tight_layout(); plt.show()
```
실행 결과 예시
```text
[1] 노선망 최대 통과능력 290대/시, 최소 절단(병목) 구간: A2, B1, C1, X1
[2] 구간별 증설 민감도(병목일수록 증설 효과가 큼)
구간 유형 용량 용량 20% 증설 시 증가(대/시)
B1 일반 100 15.0
C1 일반 70 14.0
X1 교차로 50 10.0
A2 교량 70 10.0
A1 일반 130 0.0
[3] 최종일(day 149) 사전 병목 경보: 유효 이용률 0.9 이상 구간 ['A1', 'A2']
[4] 구간 폐쇄 시나리오(수요 200대/시): 우회 경로와 평균 통과시간
시나리오 새 병목 최대유량 평균 통과(분) 주 경로 수요 대비
평시 - 290.0 40.9 A·B·C 분산 수용
A2 폐쇄 B1,C1,X1 220.0 44.9 B1->B3 수용
B2 폐쇄 A2,B3,C1 235.0 41.0 B1->B3 수용
A3 폐쇄 B3,C1 165.0 70.9 B1->B3 과부하
A2 폐쇄 시 평균 통과시간 +4.0분, 수요 200대/시 대비 최대유량 220대/시 -> 수용(과부하면 우회만으로 부족하므로 물량 조정·타 수단 검토가 필요)
[5] 파생 피처 14개 생성(lag·이용률·BPR·폭증비·요일 순환·구간 더미 등)
fold 학습 검증 BPR MAPE(%) 기준모델 MAPE(%)
1 310 280 6.72 3.00
2 590 280 6.02 2.54
3 870 280 9.16 3.93
4 1150 280 9.24 2.78
교차검증 평균 MAPE 3.06% (목표 25% 이하: 달성), 폴드가 뒤로 갈수록 학습 구간이 길어져 미래 정보 누수가 없다
피처 중요도 상위 5: util_eff 0.35, util_eff4 0.34, bpr_base 0.13, lag1_delay 0.09, km 0.02 / 중요도 1% 미만 14개(차원 축소 후보)
```
## 미션 5. 우회 수송 시 추가 소요 시간 및 유류 소모량 산출 엔진 개발
- 난이도: 중급~고급 / 예상 소요시간: 3.5시간
- 코드 설명: 우회 수송 시 추가 소요 시간과 유류 소모량을 산출하는 엔진을 구현하고, 기준 모델(BPR) 대비 앙상블 지체 예측 모델의 오차 개선율과 R2를 Hold-out 데이터로 검증한다.
```python
"""우회 수송 시 추가 소요 시간과 유류 소모량을 산출하는 엔진을 구현하고, 기준 모델(BPR) 대비 앙상블 지체 예측 모델의 오차 개선율과 R2를 Hold-out 데이터로 검증한다."""
import numpy as np
import pandas as pd
from scipy.optimize import linprog
from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor
from sklearn.metrics import r2_score
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
# ---------------- 1) 우회 엔진: 추가 소요 시간·유류 (가정값: 대당 주행 L/km, 정체 공회전 L/분) ----------------
L_KM, L_IDLE = 0.30, 0.08 # 교육용 가정 유류 계수(실제 장비 제원 아님)
def trip(res, fleet_vehicles):
"""경로별 물량 비율로 평균 통과시간·거리·유류(대당 L)를 구하고 총량으로 환산한다."""
tot = sum(res["paths"].values())
km = sum(f * sum(KM[s] for s in p) for p, f in res["paths"].items()) / tot
mins = res["avg_min"]
free = sum(f * sum(T0[s] for s in p) for p, f in res["paths"].items()) / tot
fuel = km * L_KM + max(mins - free, 0) * L_IDLE
return {"min": mins, "km": km, "fuel_l": fuel, "fleet_fuel_l": fuel * fleet_vehicles}
DEMAND, FLEET = 200, 600 # 피크 200대/시, 일 수송 600대(가정)
norm = trip(assign(DEMAND), FLEET)
rows = []
for name, closed, cm in [("A2 교량 폐쇄", ("A2",), None), ("B2 터널 폐쇄", ("B2",), None), ("A1 용량 50% 감소(정비)", (), {"A1": 0.5}), ("A2 폐쇄 + B1 정비(65%)", ("A2",), {"B1": 0.65})]:
r = assign(DEMAND, closed, cm)
t = trip(r, FLEET)
rows.append((name, t["min"] - norm["min"], t["km"] - norm["km"], t["fuel_l"] - norm["fuel_l"], t["fleet_fuel_l"] - norm["fleet_fuel_l"], max(r["util"].values())))
det = pd.DataFrame(rows, columns=["시나리오", "추가 소요(분/대)", "추가 거리(km/대)", "추가 유류(L/대)", "일 추가 유류(L)", "최대 이용률"]).round(2)
print(f"[1] 우회 엔진 결과(평시 기준 {norm['min']:.1f}분, {norm['km']:.1f}km, {norm['fuel_l']:.2f}L/대)")
print(det.to_string(index=False))
print(" 유류 계수는 교육용 가정값이다. 이용률 1.0 초과 시나리오는 우회만으로 수용이 어려워 사람의 물량 조정 판단이 필요하다.")
# ---------------- 2) 고급 모델: 기준 모델(BPR) vs 앙상블(GBR+RF), Hold-out ----------------
df = make_data(); d = add_features(df)
tr, te = d[d.day < 120], d[d.day >= 120]
gbr = GradientBoostingRegressor(n_estimators=250, max_depth=3, learning_rate=0.06, subsample=0.8, random_state=0).fit(tr[FEATS], tr["delay"])
rf = RandomForestRegressor(n_estimators=150, min_samples_leaf=3, random_state=0, n_jobs=1).fit(tr[FEATS], tr["delay"])
fac = te["t0"] * (1 + 0.08 * te["rain"])
P = {"기준(BPR)": te["t0"] + te["bpr_base"], "GBR": fac + gbr.predict(te[FEATS]), "RandomForest": fac + rf.predict(te[FEATS])}
P["앙상블(평균)"] = (P["GBR"] + P["RandomForest"]) / 2
tab = pd.DataFrame({k: {"MAE(분)": mae(te.transit, v), "RMSE(분)": float(np.sqrt(np.mean((te.transit - v) ** 2))), "MAPE(%)": mape(te.transit, v), "R2": r2_score(te.transit, v)} for k, v in P.items()}).T.round(3)
print(f"\n[2] Hold-out(day 120~149, {len(te)}행) 통과시간 예측 성능")
print(tab.to_string())
b, e = tab.loc["기준(BPR)"], tab.loc["앙상블(평균)"]
print(f" MAE 개선 {1 - e['MAE(분)'] / b['MAE(분)']:.1%}, RMSE 개선 {1 - e['RMSE(분)'] / b['RMSE(분)']:.1%} (목표 20% 이상: {'달성' if 1 - e['MAE(분)'] / b['MAE(분)'] >= .2 and 1 - e['RMSE(분)'] / b['RMSE(분)'] >= .2 else '미달'}), R2 {e['R2']:.3f} (목표 0.85 이상: {'달성' if e['R2'] >= .85 else '미달'})")
dl = te["delay"]; pdl = P["앙상블(평균)"] - fac
print(f" 참고: 지체 시간만 놓고 보면 R2 {r2_score(dl, pdl):.3f}, MAE {mae(dl, pdl):.2f}분 (기준 BPR MAE {mae(dl, te['bpr_base']):.2f}분). 통과시간 R2는 구간 간 길이 차이를 포함한다.")
# ---------------- 3) 불규칙 상황(비정형 수요) 적응력: 정비일·폭증 기간 오차 ----------------
stress = {"비계획 정비일": te["maint"] == 1, "긴급 불출일": te["urgent"] == 1, "우천일": te["rain"] == 1, "그 외": (te["maint"] == 0) & (te["urgent"] == 0) & (te["rain"] == 0)}
rr = [(k, int(m.sum()), mae(te.transit[m], P["기준(BPR)"][m]), mae(te.transit[m], P["앙상블(평균)"][m])) for k, m in stress.items() if m.sum() > 0]
print("\n[3] 조건별 MAE(분): 기준 vs 앙상블")
print(pd.DataFrame(rr, columns=["조건", "건수", "기준", "앙상블"]).round(2).to_string(index=False))
print("하이퍼파라미터(트리 250개, 깊이 3, 학습률 0.06)는 시계열 교차검증(M4) 결과를 바탕으로 정했고, 학습된 모델은 이 실행기에서는 저장하지 않고 설정만 출력한다.")
print("설정:", {"gbr": gbr.get_params()["n_estimators"], "rf": rf.get_params()["n_estimators"], "features": len(FEATS)})
```
실행 결과 예시
```text
[1] 우회 엔진 결과(평시 기준 40.9분, 31.2km, 9.63L/대)
시나리오 추가 소요(분/대) 추가 거리(km/대) 추가 유류(L/대) 일 추가 유류(L) 최대 이용률
A2 교량 폐쇄 3.96 2.38 0.81 485.68 1.0
B2 터널 폐쇄 0.05 -0.12 -0.02 -10.48 1.0
A1 용량 50% 감소(정비) 0.98 0.10 0.10 60.12 1.0
A2 폐쇄 + B1 정비(65%) 12.90 3.98 1.85 1113.00 1.3
유류 계수는 교육용 가정값이다. 이용률 1.0 초과 시나리오는 우회만으로 수용이 어려워 사람의 물량 조정 판단이 필요하다.
[2] Hold-out(day 120~149, 300행) 통과시간 예측 성능
MAE(분) RMSE(분) MAPE(%) R2
기준(BPR) 1.763 2.662 9.245 0.829
GBR 0.494 0.877 2.784 0.981
RandomForest 0.495 0.877 2.813 0.981
앙상블(평균) 0.475 0.847 2.725 0.983
MAE 개선 73.1%, RMSE 개선 68.2% (목표 20% 이상: 달성), R2 0.983 (목표 0.85 이상: 달성)
참고: 지체 시간만 놓고 보면 R2 0.846, MAE 0.47분 (기준 BPR MAE 1.52분). 통과시간 R2는 구간 간 길이 차이를 포함한다.
[3] 조건별 MAE(분): 기준 vs 앙상블
조건 건수 기준 앙상블
비계획 정비일 23 5.72 0.81
긴급 불출일 16 3.98 2.17
우천일 60 3.32 0.61
그 외 211 1.06 0.38
하이퍼파라미터(트리 250개, 깊이 3, 학습률 0.06)는 시계열 교차검증(M4) 결과를 바탕으로 정했고, 학습된 모델은 이 실행기에서는 저장하지 않고 설정만 출력한다.
설정: {'gbr': 250, 'rf': 150, 'features': 21}
```
## 미션 6. 전시 보급로 병목 혼잡도 및 우회 경로 비교 시뮬레이션 대시보드 구현
- 난이도: 중급~고급 / 예상 소요시간: 3.0시간
- 코드 설명: 보급로 병목 혼잡도와 우회 경로 비교 대시보드를 구현하고, 예측 건마다 기여도 상위 3개 요인을 자동 산출(요인군 제거 방식 설명)해 자연어 요약으로 보여 준다.
```python
"""보급로 병목 혼잡도와 우회 경로 비교 대시보드를 구현하고, 예측 건마다 기여도 상위 3개 요인을 자동 산출(요인군 제거 방식 설명)해 자연어 요약으로 보여 준다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import linprog
from sklearn.ensemble import GradientBoostingRegressor
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
df = make_data(); d = add_features(df)
tr, te = d[d.day < 120], d[d.day >= 120].copy()
gbr = GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["delay"])
te["pred"] = gbr.predict(te[FEATS])
# ---------------- 1) 설명 가능성: 요인군을 평균값으로 바꿨을 때 예측이 얼마나 변하는가 ----------------
GROUPS = {"계획 물량·이용률": ["plan_v", "util", "util_eff", "util_eff4", "bpr_base", "surge_ratio", "roll7_v"], "정비·유효 용량": ["maint", "cap_eff"],
"우천": ["rain"], "전일·전주 지체": ["lag1_delay", "lag7_delay"], "요일 패턴": ["is_weekend", "dow_sin", "dow_cos"],
"구간 특성": ["cap", "t0", "km", "k_교량", "k_터널", "k_교차로"]}
mean_row = tr[FEATS].mean()
contrib = pd.DataFrame(index=te.index)
for g, cols in GROUPS.items():
x = te[FEATS].copy()
x[cols] = mean_row[cols].values
contrib[g] = te["pred"] - gbr.predict(x) # 양수 = 이 요인이 지체를 늘림
top3 = contrib.abs().apply(lambda r: r.nlargest(3).index.tolist(), axis=1)
ok = int(sum(len(t) == 3 and len(set(t)) == 3 and np.isfinite(contrib.loc[i, t].values).all() for i, t in top3.items()))
weak = int((contrib.abs().apply(lambda r: r.nlargest(3).iloc[2], axis=1) < 0.01).sum())
print(f"[1] 예측 {len(te)}건 중 기여도 상위 3개 요인 자동 산출 {ok}건 ({ok / len(te):.0%}, 목표 100%)")
imp = contrib.abs().mean().sort_values(ascending=False)
print(f" (3번째 요인 기여가 0.01분 미만인 미미한 건 {weak}건: 순위는 산출되나 설명력은 낮음)")
print(" 요인군별 평균 기여 크기(분):", ", ".join(f"{k} {v:.2f}" for k, v in imp.items()))
def explain(i):
r, c = te.loc[i], contrib.loc[i]
parts = [f"{g}({c[g]:+.1f}분)" for g in top3[i]]
sev = "정체" if r["util_eff"] >= 0.9 else "혼잡" if r["util_eff"] >= 0.7 else "원활"
cause = {"정비·유효 용량": "비계획 정비로 통과 용량이 줄어" if r["maint"] else "유효 용량 변화로", "우천": "우천 영향으로", "계획 물량·이용률": "계획 물량이 용량에 근접해",
"전일·전주 지체": "전일 지체가 이어져", "요일 패턴": "요일 패턴 때문에", "구간 특성": "구간 특성상"}[top3[i][0]]
return f"{r['seg']} 구간 day {int(r['day'])}: {cause} 지체 {r['pred']:.1f}분 예상({sev}). 주요 요인 {', '.join(parts)}"
print("\n[2] 자연어 설명(예측 지체 상위 4건)")
for i in te["pred"].nlargest(4).index:
print(" -", explain(i))
# ---------------- 2) 대시보드 데이터: 최종일 구간별 혼잡도·경보 ----------------
last = te[te.day == te.day.max()].set_index("seg")
last["상태"] = np.where(last.util_eff >= 0.9, "정체", np.where(last.util_eff >= 0.7, "혼잡", "원활"))
print(f"\n[3] 최종일 구간 상태판: " + ", ".join(f"{s} {r.상태}({r.util_eff:.2f})" for s, r in last.iterrows()))
alert = last[last["pred"] >= 5].index.tolist()
print(f" 지체 5분 이상 예측 경보 구간: {alert or '없음'} -> 우회 검토 대상(실제 하달은 작전군수관 승인 후)")
# ---------------- 3) 우회 경로 비교: 과거 관행(축선 고정 배분) vs 증분 배정 ----------------
D = 200
cmp = []
for name, closed, cm in [("평시", (), None), ("A2 폐쇄", ("A2",), None), ("A1 정비(50%)", (), {"A1": 0.5}), ("B1 정비(65%)", (), {"B1": 0.65})]:
a, f = assign(D, closed, cm), fixed_assign(D, closed, cm)
cmp.append((name, f["avg_min"], a["avg_min"], max(f["util"].values()), max(a["util"].values())))
cm_df = pd.DataFrame(cmp, columns=["시나리오", "고정 배분(분)", "우회 배정(분)", "고정 최대 이용률", "배정 최대 이용률"]).round(2)
cm_df["단축률"] = (1 - cm_df["우회 배정(분)"] / cm_df["고정 배분(분)"]).map("{:.0%}".format)
print(f"\n[4] 우회 경로 비교(수요 {D}대/시)")
print(cm_df.to_string(index=False))
fig, ax = plt.subplots(2, 2, figsize=(13, 8))
piv = te.pivot(index="seg", columns="day", values="util_eff").iloc[:, -14:]
im = ax[0, 0].imshow(piv.values, aspect="auto", cmap="YlOrRd", vmin=0.3, vmax=1.2); ax[0, 0].set_yticks(range(len(piv))); ax[0, 0].set_yticklabels(piv.index)
ax[0, 0].set_title("1 Congestion heatmap (effective utilization, last 14 days)"); plt.colorbar(im, ax=ax[0, 0])
colors = ["tab:red" if s == "정체" else "tab:orange" if s == "혼잡" else "tab:green" for s in last["상태"]]
ax[0, 1].bar(last.index, last["pred"], color=colors); ax[0, 1].axhline(5, ls="--", c="k"); ax[0, 1].set_title("2 Predicted delay today (min), dashed = alert 5")
x = np.arange(len(cm_df)); ax[1, 0].bar(x - .2, cm_df["고정 배분(분)"], .4, label="fixed share"); ax[1, 0].bar(x + .2, cm_df["우회 배정(분)"], .4, label="detour assignment")
ax[1, 0].set_xticks(x); ax[1, 0].set_xticklabels(["normal", "A2 closed", "A1 maint", "B1 maint"]); ax[1, 0].legend(); ax[1, 0].set_title("3 Detour comparison (avg transit min)")
worst = te["pred"].idxmax(); c = contrib.loc[worst]
ax[1, 1].barh(["plan/util", "maint/cap", "rain", "lag delay", "weekday", "segment"], c[list(GROUPS)].values); ax[1, 1].set_title("4 Contribution for worst case (min)")
plt.suptitle("MOV01 bottleneck dashboard (synthetic)"); plt.tight_layout(); plt.show()
```
실행 결과 예시
```text
[1] 예측 300건 중 기여도 상위 3개 요인 자동 산출 300건 (100%, 목표 100%)
(3번째 요인 기여가 0.01분 미만인 미미한 건 7건: 순위는 산출되나 설명력은 낮음)
요인군별 평균 기여 크기(분): 계획 물량·이용률 0.95, 전일·전주 지체 0.41, 구간 특성 0.15, 요일 패턴 0.06, 정비·유효 용량 0.04, 우천 0.02
[2] 자연어 설명(예측 지체 상위 4건)
- C2 구간 day 134: 계획 물량이 용량에 근접해 지체 12.4분 예상(정체). 주요 요인 계획 물량·이용률(+9.1분), 구간 특성(+4.0분), 정비·유효 용량(+2.0분)
- B1 구간 day 120: 계획 물량이 용량에 근접해 지체 10.9분 예상(정체). 주요 요인 계획 물량·이용률(+8.8분), 정비·유효 용량(+0.9분), 구간 특성(+0.4분)
- A1 구간 day 142: 계획 물량이 용량에 근접해 지체 10.5분 예상(정체). 주요 요인 계획 물량·이용률(+7.5분), 정비·유효 용량(+0.7분), 전일·전주 지체(-0.1분)
- A1 구간 day 137: 계획 물량이 용량에 근접해 지체 10.3분 예상(정체). 주요 요인 계획 물량·이용률(+7.7분), 정비·유효 용량(+0.7분), 우천(+0.2분)
[3] 최종일 구간 상태판: A1 정체(0.97), A2 정체(0.97), A3 혼잡(0.82), B1 혼잡(0.89), B2 혼잡(0.83), B3 혼잡(0.81), C1 원활(0.60), C2 원활(0.69), X1 원활(0.61), X2 원활(0.46)
지체 5분 이상 예측 경보 구간: 없음 -> 우회 검토 대상(실제 하달은 작전군수관 승인 후)
[4] 우회 경로 비교(수요 200대/시)
시나리오 고정 배분(분) 우회 배정(분) 고정 최대 이용률 배정 최대 이용률 단축률
평시 41.91 40.91 0.91 1.0 2%
A2 폐쇄 91.40 44.87 1.24 1.0 51%
A1 정비(50%) 42.64 41.89 0.98 1.0 2%
B1 정비(65%) 53.08 42.14 1.23 1.0 21%
```
## 미션 7. 우회로 수송 명령 하달을 위한 작전군수관 HITL 통제 워크플로우 설계
- 난이도: 고급 / 예상 소요시간: 3.5시간
- 코드 설명: 선형계획(LP)으로 우회 수송 물량 배분을 최적화해 과거 관행 대비 비용 절감을 백테스트하고, 우회 수송 명령을 작전군수관이 직접 승인·반려하는 HITL 결재 워크플로우(권한·2단계 결재·해시 체인 감사 로그)를 구현한다.
```python
"""선형계획(LP)으로 우회 수송 물량 배분을 최적화해 과거 관행 대비 비용 절감을 백테스트하고, 우회 수송 명령을 작전군수관이 직접 승인·반려하는 HITL 결재 워크플로우(권한·2단계 결재·해시 체인 감사 로그)를 구현한다."""
import hashlib
import json
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import linprog
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
# ---------------- 1) 최적화 시뮬레이터: 과거 관행(축선 고정 배분) vs LP ----------------
SC = [("평시", 200, (), None), ("수송량 폭증(+30%)", 260, (), None), ("A2 교량 정비(용량 65%)", 200, (), {"A2": .65}), ("A2 교량 폐쇄", 200, ("A2",), None),
("A2 폐쇄 + 우천", 200, ("A2",), {k: .92 for k in CAP}), ("B1 정비 + 폭증", 250, (), {"B1": .65})]
rows, plans = [], {}
for name, D, cl, cm in SC:
f, l = fixed_assign(D, cl, cm), lp_assign(D, cl, cm)
cf, cl_ = cost_of(f), cost_of(l)
rows.append((name, D, f["avg_min"], l["avg_min"], cf / 1e4, cl_ / 1e4, 1 - cl_ / cf))
plans[name] = (D, cl, cm, l, cf, cl_)
res = pd.DataFrame(rows, columns=["시나리오", "수요", "고정 배분(분)", "LP(분)", "고정 비용(만원/시)", "LP 비용(만원/시)", "절감률"])
print("[1] 시나리오별 비용 비교(시간·유류 단가는 교육용 가정값)")
print(res.assign(절감률=res["절감률"].map("{:.1%}".format)).round(1).to_string(index=False))
avg = res["절감률"].mean()
print(f" 시나리오 6종 평균 절감 {avg:.1%}, 우회가 필요한 4개 시나리오 평균 {res['절감률'].iloc[2:].mean():.1%} (목표 15% 이상: {'달성' if avg >= .15 else '미달'}), 평시는 {res['절감률'].iloc[0]:.1%}에 그친다")
# ---------------- 2) 백테스팅: 과거 30일(day 120~149) 실적 조건으로 재현 ----------------
df = make_data()
rec = []
for day, g in df[df.day >= 120].groupby("day"):
D = float(g[g.seg.isin(["A1", "B1", "C1"])]["plan_v"].sum())
cm = {r.seg: r.cap_eff / r.cap for r in g.itertuples() if r.cap_eff < r.cap}
f, l = fixed_assign(D, (), cm), lp_assign(D, (), cm)
rec.append((day, D, cost_of(f), cost_of(l)))
bt = pd.DataFrame(rec, columns=["day", "D", "fix", "lp"])
bt["save"] = 1 - bt["lp"] / bt["fix"]
print(f"\n[2] 백테스팅(30일): 평균 절감 {bt['save'].mean():.1%}(목표 15% 이상: {'달성' if bt['save'].mean() >= .15 else '미달'}), 최대 {bt['save'].max():.1%}(day {int(bt.loc[bt['save'].idxmax(), 'day'])}), 절감 0 이하인 날 {int((bt['save'] <= 0).sum())}일")
print(f" 절감 10% 미만인 날 {int((bt['save'] < .10).sum())}일: 효과는 정비·폭증이 겹친 날에 집중된다. 과거 관행 배분은 가정이므로 실제 부대 실적으로는 별도 검증이 필요하다.")
# ---------------- 3) 민감도: 돌발 수요(전시 전환 가정) ----------------
sens = []
for k in (0.8, 1.0, 1.2, 1.4, 1.6):
f, l = fixed_assign(200 * k, ("A2",)), lp_assign(200 * k, ("A2",))
sens.append((f"x{k}", 200 * k, max(l["util"].values()), 1 - cost_of(l) / cost_of(f)))
sdf = pd.DataFrame(sens, columns=["수요 배율", "수요", "LP 최대 이용률", "절감률"])
print("[3] 돌발 수요 민감도(A2 폐쇄 상황):", "; ".join(f"{r['수요 배율']} -> 이용률 {r['LP 최대 이용률']:.2f}, 절감 {r['절감률']:.0%}" for _, r in sdf.iterrows()))
print(" 수요 x1.2 이상에서는 최대유량(220대/시)을 넘어 우회만으로 수용할 수 없으므로 물량 조정 결정이 필요하다.")
# ---------------- 4) HITL 결재 워크플로우 ----------------
USERS = {"plan_a": "PLN", "plan_b": "PLN", "ops_kim": "OPS", "ops_lee": "OPS", "sup_park": "SUP", "aud_han": "AUD"}
PERM = {"PLN": {"submit", "redraft"}, "OPS": {"approve", "reject"}, "SUP": {"approve", "reject"}, "AUD": {"read"}}
class Flow:
def __init__(self):
self.docs, self.log, self.blocked = {}, [], []
def _audit(self, user, action, doc, detail):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
rec = {"seq": len(self.log) + 1, "user": user, "action": action, "doc": doc, "detail": detail, "prev": prev}
rec["hash"] = hashlib.sha256(json.dumps(rec, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:12]
self.log.append(rec)
def verify(self):
prev = "GENESIS"
for r in self.log:
body = {k: v for k, v in r.items() if k != "hash"}
if r["prev"] != prev or hashlib.sha256(json.dumps(body, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:12] != r["hash"]:
return False, r["seq"]
prev = r["hash"]
return True, None
def deny(self, user, act, doc, why):
self.blocked.append((user, act, doc, why)); self._audit(user, "DENIED:" + act, doc, why); return False
def submit(self, user, scen, parent=None):
need = "redraft" if parent else "submit"
if need not in PERM.get(USERS.get(user), set()):
return self.deny(user, need, "-", "상신 권한 없음")
D, cl, cm, l, cf, cl_ = plans[scen]
extra = l["avg_min"] - plans["평시"][3]["avg_min"]
big = extra >= 3 or max(l["util"].values()) > 0.98 # 추가 소요 3분 이상 또는 용량 포화 -> 2단계 결재
did = f"ORD{len(self.docs) + 1:02d}"
self.docs[did] = {"scen": scen, "line": ["OPS", "SUP"] if big else ["OPS"], "step": 0, "status": "검토대기", "drafter": user, "done": [], "parent": parent}
self._audit(user, "submit", did, f"{scen} 추가소요 {extra:+.1f}분")
return did
def decide(self, user, did, act, reason=""):
role, d = USERS.get(user), self.docs.get(did)
if role is None: return self.deny(user, act, did, "미등록 사용자")
if d is None or d["status"] != "검토대기": return self.deny(user, act, did, "결재 대상 아님(처리 완료)")
if act not in PERM[role]: return self.deny(user, act, did, f"{role} 역할은 {act} 불가")
if d["line"][d["step"]] != role: return self.deny(user, act, did, "결재 순서 위반")
if user == d["drafter"] or user in d["done"]: return self.deny(user, act, did, "기안자·기결재자 재결재 금지")
if len(reason.strip()) < 10: return self.deny(user, act, did, "사유 10자 이상 필수")
if act == "reject":
d["status"] = "반려"
else:
d["done"].append(user); d["step"] += 1
d["status"] = "승인" if d["step"] == len(d["line"]) else "검토대기"
self._audit(user, act, did, reason); return True
def edit(self, user, did):
if self.docs[did]["status"] == "승인": return self.deny(user, "edit", did, "승인 후 명령 변경 금지(재기안)")
return True
wf = Flow()
names = ["A2 교량 폐쇄", "A2 교량 정비(용량 65%)", "수송량 폭증(+30%)", "B1 정비 + 폭증", "A2 폐쇄 + 우천", "평시"]
ids = [wf.submit("plan_a" if i % 2 == 0 else "plan_b", n) for i, n in enumerate(names)]
print(f"\n[4] AI 우회안 {len(ids)}건 전부 '검토대기' 등록, 자동 승인 {sum(wf.docs[i]['status'] == '승인' for i in ids)}건 (목표 0건). 2단계 결재 대상 {[i for i in ids if len(wf.docs[i]['line']) == 2]}")
wf.decide("ops_kim", ids[0], "reject", "추가 소요와 용량 포화 우려, 물량 조정안 포함해 재기안")
new = wf.submit("plan_a", names[0], parent=ids[0])
wf.decide("ops_kim", new, "approve", "우회 경로 용량 확인, 상급 승인 상신함")
wf.decide("sup_park", new, "approve", "우회 시행, 운행 중 지체 보고 조건")
wf.decide("ops_lee", ids[1], "approve", "정비 구간 용량 감소분 우회 반영 타당")
wf.decide("ops_kim", ids[5], "approve", "평시 물량으로 변경 사유 없어 기존 유지")
attacks = [("plan_a", ids[2], "approve", "내가 올린 안을 직접 승인한다", "요청자 자기 승인"), ("aud_han", ids[2], "approve", "감사 계정으로 승인 시도", "감사 역할 승인"),
("ghost", ids[2], "approve", "외부 계정으로 승인 시도", "미등록 사용자"), ("ops_kim", ids[2], "approve", "", "사유 없는 승인"),
("ops_kim", ids[2], "reject", "짧음", "사유 짧은 반려"), ("sup_park", ids[2], "approve", "상급자가 순서를 건너뛰어 승인", "결재 순서 건너뛰기"),
("ops_kim", ids[5], "approve", "이미 승인된 문서를 다시 승인", "처리 완료 재승인"), ("sup_park", new, "reject", "승인된 문서를 뒤집어 반려", "승인 후 번복"),
("plan_b", ids[3], "approve", "요청자 계정으로 승인 시도함", "요청자 권한 위반")]
for u, doc, act, rs, _ in attacks:
wf.decide(u, doc, act, rs)
wf.edit("plan_a", new) # 10) 승인 후 명령 변경
print(f" 권한 우회 시도 10건 중 차단 {len(wf.blocked)}건 (차단율 {len(wf.blocked) / 10:.0%}, 목표 100%)")
for u, a, doc, why in wf.blocked[:4]:
print(f" 차단 {u:<9} {a:<8} {doc}: {why}")
ok, _ = wf.verify()
snap = json.loads(json.dumps(wf.log))
wf.log[3]["detail"] = "사후에 고쳐 쓴 사유"
ok2, at = wf.verify()
wf.log = snap
print(f" 감사 로그 {len(wf.log)}건 해시 체인 무결성 {'100%' if ok else '훼손'}; 4번 기록 사후 수정 시 {'통과' if ok2 else f'위변조 검출({at}번)'}")
print(f" {ids[0]}(반려) -> {new}(재기안, parent 연결) 이력:", [(r['action'], r['user']) for r in wf.log if r['doc'] in (ids[0], new)])
print(" 문서 상태:", pd.Series({k: v["status"] for k, v in wf.docs.items()}).value_counts().to_dict(), "| 실제 우회 명령 하달은 지휘계통(작전군수관 승인) 결심에 따른다.")
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
x = np.arange(len(res)); ax[0].bar(x - .2, res["고정 비용(만원/시)"], .4, label="fixed share"); ax[0].bar(x + .2, res["LP 비용(만원/시)"], .4, label="LP")
ax[0].set_xticks(x); ax[0].set_xticklabels([f"S{i + 1}" for i in x]); ax[0].legend(); ax[0].set_title("Cost per hour by scenario")
ax[1].plot(sdf["수요"], sdf["절감률"] * 100, "o-"); ax[1].set_title("Sensitivity: demand vs saving (%), A2 closed"); ax[1].set_xlabel("demand (veh/h)")
plt.tight_layout(); plt.show()
```
실행 결과 예시
```text
[1] 시나리오별 비용 비교(시간·유류 단가는 교육용 가정값)
시나리오 수요 고정 배분(분) LP(분) 고정 비용(만원/시) LP 비용(만원/시) 절감률
평시 200 41.9 41.8 1124.8 1122.7 0.2%
수송량 폭증(+30%) 260 57.2 44.8 1860.5 1547.9 16.8%
A2 교량 정비(용량 65%) 200 53.3 42.2 1351.6 1138.9 15.7%
A2 교량 폐쇄 200 91.4 45.1 2122.3 1207.2 43.1%
A2 폐쇄 + 우천 200 119.7 47.0 2688.8 1245.4 53.7%
B1 정비 + 폭증 250 76.5 46.2 2270.9 1523.7 32.9%
시나리오 6종 평균 절감 27.1%, 우회가 필요한 4개 시나리오 평균 36.4% (목표 15% 이상: 달성), 평시는 0.2%에 그친다
[2] 백테스팅(30일): 평균 절감 16.0%(목표 15% 이상: 달성), 최대 48.5%(day 134), 절감 0 이하인 날 0일
절감 10% 미만인 날 12일: 효과는 정비·폭증이 겹친 날에 집중된다. 과거 관행 배분은 가정이므로 실제 부대 실적으로는 별도 검증이 필요하다.
[3] 돌발 수요 민감도(A2 폐쇄 상황): x0.8 -> 이용률 0.85, 절감 4%; x1.0 -> 이용률 0.95, 절감 43%; x1.2 -> 이용률 1.32, 절감 58%; x1.4 -> 이용률 1.54, 절감 58%; x1.6 -> 이용률 1.76, 절감 56%
수요 x1.2 이상에서는 최대유량(220대/시)을 넘어 우회만으로 수용할 수 없으므로 물량 조정 결정이 필요하다.
[4] AI 우회안 6건 전부 '검토대기' 등록, 자동 승인 0건 (목표 0건). 2단계 결재 대상 ['ORD01', 'ORD03', 'ORD04', 'ORD05']
권한 우회 시도 10건 중 차단 10건 (차단율 100%, 목표 100%)
차단 plan_a approve ORD03: PLN 역할은 approve 불가
차단 aud_han approve ORD03: AUD 역할은 approve 불가
차단 ghost approve ORD03: 미등록 사용자
차단 ops_kim approve ORD03: 사유 10자 이상 필수
감사 로그 22건 해시 체인 무결성 100%; 4번 기록 사후 수정 시 위변조 검출(4번)
ORD01(반려) -> ORD07(재기안, parent 연결) 이력: [('submit', 'plan_a'), ('reject', 'ops_kim'), ('submit', 'plan_a'), ('approve', 'ops_kim'), ('approve', 'sup_park'), ('DENIED:reject', 'sup_park'), ('DENIED:edit', 'plan_a')]
문서 상태: {'승인': 3, '검토대기': 3, '반려': 1} | 실제 우회 명령 하달은 지휘계통(작전군수관 승인) 결심에 따른다.
```
## 미션 8. 주요 축선 3개 동시 봉쇄 상황 모의 우회 수송 시뮬레이션 스트레스 테스트
- 난이도: 고급 / 예상 소요시간: 3.0시간
- 코드 설명: 가상 노선망에서 구간 3개가 동시에 막히는 모든 조합(120가지)과 축선 3개 전면 봉쇄를 모의해 단절·과부하·추가 지체를 평가하고, 대시보드 갱신 시간(2초 이내)과 지체 경보 정확도(95% 이상)를 스트레스 조건에서 측정한다.
```python
"""가상 노선망에서 구간 3개가 동시에 막히는 모든 조합(120가지)과 축선 3개 전면 봉쇄를 모의해 단절·과부하·추가 지체를 평가하고, 대시보드 갱신 시간(2초 이내)과 지체 경보 정확도(95% 이상)를 스트레스 조건에서 측정한다."""
import itertools
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import linprog
from sklearn.ensemble import GradientBoostingRegressor
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
D = 150
base = assign(D)
print(f"[0] 평시: 수요 {D}대/시, 평균 통과 {base['avg_min']:.1f}분, 최대유량 {max_flow()[0]:.0f}대/시")
# ---------------- 1) 축선 3개(A·B·C) 전면 봉쇄 ----------------
axes_cut = ("A3", "B3", "C2") # 3개 축선의 목적지 접근 구간
mf, _ = max_flow(axes_cut)
print(f"[1] 3개 축선 접근 구간(A3·B3·C2) 동시 봉쇄: 최대유량 {mf:.0f}대/시, 경로 {len(PATHS) - len([p for p in PATHS if not any(s in axes_cut for s in p)])}/{len(PATHS)}개 차단 -> "
+ ("도로 수송 불가. 공중·해상 등 타 수단 또는 물량 보류 결정이 필요하며 사람이 판단한다" if assign(D, axes_cut) is None else "우회 가능"))
# ---------------- 2) 구간 3개 동시 봉쇄 120조합 ----------------
rows = []
t0 = time.perf_counter()
for c in itertools.combinations(CAP, 3):
flow, cut = max_flow(c)
r = assign(D, c) if flow > 0 else None
rows.append({"closed": "+".join(c), "maxflow": flow, "avg_min": r["avg_min"] if r else np.nan, "max_util": max(r["util"].values()) if r else np.nan})
el = time.perf_counter() - t0
st = pd.DataFrame(rows)
st["상태"] = np.where(st["maxflow"] <= 0, "단절", np.where(st["maxflow"] < D, "수요 초과", np.where(st["avg_min"] > base["avg_min"] * 1.5, "심각 지체", "수용")))
vc = st["상태"].value_counts()
print(f"\n[2] 120개 조합 평가({el:.2f}초): " + ", ".join(f"{k} {v}" for k, v in vc.items()))
ok_rows = st[st["상태"].isin(["수용", "심각 지체"])]
print(f" 수용 가능 조합의 평균 통과시간 {ok_rows['avg_min'].mean():.1f}분(평시 {base['avg_min']:.1f}분), 최악 수용 조합 {ok_rows.loc[ok_rows['avg_min'].idxmax(), 'closed']} {ok_rows['avg_min'].max():.1f}분")
cnt = pd.Series([s for c in st[st["상태"] == "단절"]["closed"] for s in c.split("+")]).value_counts()
print(f" 단절 조합에 자주 포함된 구간: {', '.join(f'{k}({v})' for k, v in cnt.head(4).items())} -> 이중화(우회로 확보) 우선 후보")
# ---------------- 3) 대시보드 갱신 부하: 구간 예측 + 최대유량 + 배정을 반복 ----------------
df = make_data(); d = add_features(df)
tr, te = d[d.day < 120], d[d.day >= 120].copy()
gbr = GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["delay"])
def refresh(day, closed=()):
x = te[te.day == day]
pred = gbr.predict(x[FEATS])
flow, cut = max_flow(closed)
r = assign(float(x[x.seg.isin(["A1", "B1", "C1"])]["plan_v"].sum()), closed)
return pred, flow, cut, r
lat = []
rng = np.random.default_rng(8)
combos = list(itertools.combinations(CAP, 2))
for k in range(60):
c = combos[int(rng.integers(len(combos)))]
t1 = time.perf_counter(); refresh(120 + k % 30, c); lat.append(time.perf_counter() - t1)
lat = np.array(lat)
print(f"\n[3] 대시보드 데이터 갱신 60회(구간 2개 폐쇄 무작위): 평균 {lat.mean() * 1000:.0f}ms, 최대 {lat.max() * 1000:.0f}ms (목표 2초 이내: {'달성' if lat.max() < 2 else '미달'}; 화면 렌더링 시간은 제외)")
# ---------------- 4) 경보 정확도: 예측 지체 5분 이상 경보 vs 실제 5분 이상 ----------------
te["pred"] = gbr.predict(te[FEATS])
a_pred, a_true = te["pred"] >= 5, te["delay"] >= 5
acc = float((a_pred == a_true).mean())
tp, fp, fn = int((a_pred & a_true).sum()), int((a_pred & ~a_true).sum()), int((~a_pred & a_true).sum())
print(f"[4] 경보 정확도: {acc:.1%} (목표 95% 이상: {'달성' if acc >= .95 else '미달'}), 실제 지체 5분 이상 {int(a_true.sum())}건 중 탐지 {tp}건(재현율 {tp / max(tp + fn, 1):.0%}), 오경보 {fp}건")
print(f" 경보 기준을 3분으로 낮추면: ", end="")
a3 = te["pred"] >= 3; t3 = te["delay"] >= 5
print(f"재현율 {float((a3 & t3).sum() / max(t3.sum(), 1)):.0%}, 오경보 {int((a3 & ~t3).sum())}건 -> 놓치지 않는 쪽이 중요하면 기준을 낮추되 오경보 피로를 함께 본다")
print("모든 모의는 교육용 가상 노선이며, 실제 봉쇄·우회 결심은 지휘계통(작전군수관 승인)에 따른다.")
fig, ax = plt.subplots(1, 3, figsize=(15, 4))
vc.rename(index={"단절": "cut off", "수요 초과": "over demand", "수용": "ok", "심각 지체": "severe delay"}).plot.bar(ax=ax[0], color="tab:blue")
ax[0].set_title("120 triple-closure outcomes"); ax[0].set_xlabel(""); ax[0].tick_params(axis="x", rotation=15)
ax[1].hist(ok_rows["avg_min"], bins=15); ax[1].axvline(base["avg_min"], c="r"); ax[1].set_title("Avg transit (min) of feasible combos")
ax[2].hist(lat * 1000, bins=15); ax[2].set_title("Refresh latency (ms)")
plt.tight_layout(); plt.show()
```
실행 결과 예시
```text
[0] 평시: 수요 150대/시, 평균 통과 37.8분, 최대유량 290대/시
[1] 3개 축선 접근 구간(A3·B3·C2) 동시 봉쇄: 최대유량 0대/시, 경로 8/8개 차단 -> 도로 수송 불가. 공중·해상 등 타 수단 또는 물량 보류 결정이 필요하며 사람이 판단한다
[2] 120개 조합 평가(0.12초): 수요 초과 73, 수용 44, 단절 3
수용 가능 조합의 평균 통과시간 43.3분(평시 37.8분), 최악 수용 조합 A1+A2+B3 52.5분
단절 조합에 자주 포함된 구간: C1(2), A3(2), B3(2), A1(1) -> 이중화(우회로 확보) 우선 후보
[3] 대시보드 데이터 갱신 60회(구간 2개 폐쇄 무작위): 평균 7ms, 최대 10ms (목표 2초 이내: 달성; 화면 렌더링 시간은 제외)
[4] 경보 정확도: 97.0% (목표 95% 이상: 달성), 실제 지체 5분 이상 22건 중 탐지 15건(재현율 68%), 오경보 2건
경보 기준을 3분으로 낮추면: 재현율 91%, 오경보 46건 -> 놓치지 않는 쪽이 중요하면 기준을 낮추되 오경보 피로를 함께 본다
모든 모의는 교육용 가상 노선이며, 실제 봉쇄·우회 결심은 지휘계통(작전군수관 승인)에 따른다.
```
## 미션 9. AAR 기반 병목 예측 오차 분석 및 도로 통과 용량 파라미터 재보정
- 난이도: 고급 / 예상 소요시간: 3.0시간
- 코드 설명: AAR 관점에서 BPR 기준 모델의 병목 예측 오차를 원인별로 분석하고, 최소제곱으로 구간별 유효 통과 용량 배율과 지체 함수 파라미터를 재보정해 검증 구간 오차를 줄인 뒤, 효과를 비용 절감액·시간 단축률·경보 포착률 3대 지표로 환산하고 민감도를 분석한다.
```python
"""AAR 관점에서 BPR 기준 모델의 병목 예측 오차를 원인별로 분석하고, 최소제곱으로 구간별 유효 통과 용량 배율과 지체 함수 파라미터를 재보정해 검증 구간 오차를 줄인 뒤, 효과를 비용 절감액·시간 단축률·경보 포착률 3대 지표로 환산하고 민감도를 분석한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import least_squares, linprog
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
df = make_data(); d = add_features(df)
SEGL = list(SEGS["seg"])
cal, val = d[d.day < 135], d[d.day >= 135] # AAR 대상(보정) 구간 / 검증 구간
def model(p, x):
"""보정 모델: a, b, c(전일 지체 이월), 구간별 용량 배율 m_s, 정비 시 용량 배율, 우천 시 용량 배율."""
a, b, c, mm, rr = p[0], p[1], p[2], p[3], p[4]
m = pd.Series(p[5:], index=SEGL)[x["seg"]].values
eff = x["cap"].values * m * np.where(x["maint"] == 1, mm, 1.0) * np.where(x["rain"] == 1, rr, 1.0)
return x["t0"].values * a * (x["plan_v"].values / eff) ** b + c * x["lag1_delay"].values
p0 = np.r_[0.15, 4.0, 0.0, 1.0, 1.0, np.ones(len(SEGL))]
lo = np.r_[0.01, 1.0, 0.0, 0.3, 0.5, np.full(len(SEGL), 0.5)]
hi = np.r_[1.0, 6.0, 0.8, 1.2, 1.2, np.full(len(SEGL), 1.5)]
fit = least_squares(lambda p: model(p, cal) - cal["delay"].values, p0, bounds=(lo, hi))
P = fit.x
# ---------------- 1) 오차 분석(AAR) ----------------
base_cal, base_val = cal["bpr_base"].values, val["bpr_base"].values
new_val = model(P, val)
print(f"[1] 검증 구간(day 135~149, {len(val)}행) 지체 MAE: 기준 {mae(val.delay, base_val):.2f}분 -> 재보정 {mae(val.delay, new_val):.2f}분 ({1 - mae(val.delay, new_val) / mae(val.delay, base_val):.0%} 개선)")
print(f" 통과시간 MAPE: 기준 {mape(val.transit, val.t0 + base_val):.2f}% -> 재보정 {mape(val.transit, val.t0 * (1 + .08 * val.rain) + new_val):.2f}%")
err = pd.DataFrame({"원인": ["비계획 정비일", "긴급 불출일", "우천일", "정상일"],
"건수": [int((cal.maint == 1).sum()), int((cal.urgent == 1).sum()), int((cal.rain == 1).sum()), int(((cal.maint + cal.urgent + cal.rain) == 0).sum())],
"기준 평균오차": [(base_cal - cal.delay)[m].mean() for m in (cal.maint == 1, cal.urgent == 1, cal.rain == 1, (cal.maint + cal.urgent + cal.rain) == 0)],
"재보정 평균오차": [(model(P, cal) - cal.delay)[m].mean() for m in (cal.maint == 1, cal.urgent == 1, cal.rain == 1, (cal.maint + cal.urgent + cal.rain) == 0)]}).round(2)
print("[2] AAR 원인별 평균 오차(예측-실제, 분; 음수 = 과소예측)")
print(err.to_string(index=False))
pm = pd.DataFrame({"파라미터": ["a(지체 계수)", "b(비선형 지수)", "c(전일 이월)", "정비 시 용량 배율", "우천 시 용량 배율"], "공칭/기준": [0.15, 4.0, 0.0, 1.0, 1.0], "재보정": P[:5]}).round(3)
print("[3] 재보정 파라미터:", "; ".join(f"{r['파라미터']} {r['공칭/기준']} -> {r['재보정']}" for _, r in pm.iterrows()))
ms = pd.Series(P[5:], index=SEGL).round(2)
ms_txt = {k: float(v) for k, v in ms.items()}
print(" 구간별 유효 용량 배율:", ms_txt, "(1.0 = 공칭 용량과 일치)")
# ---------------- 2) 효과 환산: 검증 15일 백테스트 ----------------
TV = 300 # 대당 분당 시간가치(원, 가정)
hours_per_day, APPLY = 2, 0.4 # 일 피크 2시간, 승인·현장 제약으로 실제 적용되는 비율 40%(가정)
rec = []
for day, g in df[df.day >= 135].groupby("day"):
D = float(g[g.seg.isin(["A1", "B1", "C1"])]["plan_v"].sum())
cm = {r.seg: r.cap_eff / r.cap for r in g.itertuples() if r.cap_eff < r.cap}
f, l = fixed_assign(D, (), cm), lp_assign(D, (), cm)
pr_b, pr_n = val[val.day == day], val[val.day == day]
rec.append((day, cost_of(f, tv=TV), cost_of(l, tv=TV), f["avg_min"], l["avg_min"], D, float(g.delay.max()),
float((pr_b.bpr_base).max()), float(model(P, pr_n).max())))
bt = pd.DataFrame(rec, columns=["day", "c_fix", "c_lp", "t_fix", "t_lp", "D", "max_delay", "pred_base", "pred_new"])
saving_day = (bt["c_fix"] - bt["c_lp"]) * hours_per_day * APPLY
ann = saving_day.mean() * 365
time_cut = 1 - bt["t_lp"].sum() / bt["t_fix"].sum()
veh_min = float(((bt["t_fix"] - bt["t_lp"]) * bt["D"] * hours_per_day * APPLY).sum())
alert_true = bt["max_delay"] >= 5
cap_b = float(((bt["pred_base"] >= 5) & alert_true).sum() / max(alert_true.sum(), 1)); cap_n = float(((bt["pred_new"] >= 5) & alert_true).sum() / max(alert_true.sum(), 1))
fa_b, fa_n = int(((bt["pred_base"] >= 5) & ~alert_true).sum()), int(((bt["pred_new"] >= 5) & ~alert_true).sum())
print(f"\n[4] 3대 정량 효과 지표(검증 15일, 고정 배분 대비 LP 우회 최적화, 가정 단가 사용)")
print(f" (1) 비용 절감액: 일 평균 {saving_day.mean() / 1e4:,.0f}만원, 연 환산 {ann / 1e8:,.1f}억원(가정: 일 피크 2시간, 적용률 40%, 365일)")
print(f" (2) 시간 단축률: 평균 통과시간 {bt['t_fix'].mean():.1f}분 -> {bt['t_lp'].mean():.1f}분 ({time_cut:.1%}), 절감 차량·시간 {veh_min / 60:,.0f}대·시간")
print(f" (3) 지체 경보 포착률(실제 지체 5분 이상 {int(alert_true.sum())}일 대비): 기준 {cap_b:.0%} -> 재보정 {cap_n:.0%} (오경보 {fa_b} -> {fa_n}일)")
# ---------------- 3) ROI와 민감도(가정 도입·운영비) ----------------
COST = 2.0e8 # 도입·운영 연 비용 2억원(교육용 가정)
roi = (ann - COST) / COST
print(f"\n[5] ROI = (연 절감 - 비용) / 비용 = {roi:.1%} (연 비용 {COST / 1e8:.1f}억원 가정), 회수 기간 {COST / ann * 12:.1f}개월")
sens = pd.DataFrame({"시간가치 -30%": [ann * 0.7], "기준": [ann], "시간가치 +30%": [ann * 1.3]}).T[0]
low_eff = ann * 0.5
print(" 민감도(연 절감, 억원):", {k: round(float(v) / 1e8, 1) for k, v in sens.items()}, f"| 효과가 절반이면 ROI {(low_eff - COST) / COST:.1%}")
print(" 단계별 로드맵: 단기(1~3개월) 경보 기준 정착 -> 중기(6개월) 구간별 용량 배율 월 단위 재보정 -> 장기(1년) 상위 정보체계 연동 검토")
print(" 추정 산식은 가정 단가에 크게 의존하므로 실제 적용 때는 기획·재정 부서가 단가를 제시해야 하며 본 값은 학습용이다.")
fig, ax = plt.subplots(1, 3, figsize=(15, 4))
ax[0].scatter(val["delay"], base_val, s=8, label="baseline"); ax[0].scatter(val["delay"], new_val, s=8, label="recalibrated"); ax[0].plot([0, 15], [0, 15], "k--"); ax[0].legend(); ax[0].set_title("Actual vs predicted delay (min)")
ax[1].bar(ms.index, ms.values); ax[1].axhline(1, c="k", ls="--"); ax[1].set_title("Effective capacity multiplier by segment")
ax[2].bar(["-30%", "base", "+30%"], [v / 1e8 for v in sens.values]); ax[2].set_title("Annual saving sensitivity (100M won)")
plt.tight_layout(); plt.show()
```
실행 결과 예시
```text
[1] 검증 구간(day 135~149, 150행) 지체 MAE: 기준 1.69분 -> 재보정 0.54분 (68% 개선)
통과시간 MAPE: 기준 10.02% -> 재보정 3.05%
[2] AAR 원인별 평균 오차(예측-실제, 분; 음수 = 과소예측)
원인 건수 기준 평균오차 재보정 평균오차
비계획 정비일 78 -4.73 -0.03
긴급 불출일 69 -2.40 -1.33
우천일 160 -1.55 0.01
정상일 994 -0.81 0.09
[3] 재보정 파라미터: a(지체 계수) 0.15 -> 0.197; b(비선형 지수) 4.0 -> 3.091; c(전일 이월) 0.0 -> 0.258; 정비 시 용량 배율 1.0 -> 0.623; 우천 시 용량 배율 1.0 -> 0.922
구간별 유효 용량 배율: {'A1': 0.98, 'A2': 0.98, 'A3': 1.04, 'B1': 1.01, 'B2': 0.98, 'B3': 1.04, 'C1': 1.0, 'C2': 1.0, 'X1': 0.97, 'X2': 1.03} (1.0 = 공칭 용량과 일치)
[4] 3대 정량 효과 지표(검증 15일, 고정 배분 대비 LP 우회 최적화, 가정 단가 사용)
(1) 비용 절감액: 일 평균 90만원, 연 환산 3.3억원(가정: 일 피크 2시간, 적용률 40%, 365일)
(2) 시간 단축률: 평균 통과시간 63.3분 -> 47.7분 (24.6%), 절감 차량·시간 816대·시간
(3) 지체 경보 포착률(실제 지체 5분 이상 7일 대비): 기준 0% -> 재보정 86% (오경보 0 -> 0일)
[5] ROI = (연 절감 - 비용) / 비용 = 64.9% (연 비용 2.0억원 가정), 회수 기간 7.3개월
민감도(연 절감, 억원): {'시간가치 -30%': 2.3, '기준': 3.3, '시간가치 +30%': 4.3} | 효과가 절반이면 ROI -17.5%
단계별 로드맵: 단기(1~3개월) 경보 기준 정착 -> 중기(6개월) 구간별 용량 배율 월 단위 재보정 -> 장기(1년) 상위 정보체계 연동 검토
추정 산식은 가정 단가에 크게 의존하므로 실제 적용 때는 기획·재정 부서가 단가를 제시해야 하며 본 값은 학습용이다.
```
## 미션 10. 전시 수송 병목 구간 식별 및 우회 수송 시뮬레이션 도구 통합 패키지 완성
- 난이도: 최고급 / 예상 소요시간: 3.5시간
- 코드 설명: 데이터 갱신, 정합성 검증, 드리프트 점검, 자동 재학습(챔피언/챌린저), 7일 수송 소요 예측, 병목·우회 배분, 승인대기 분리까지 한 번에 실행하는 통합 파이프라인을 완성하고 운영 문서 완비율·자체 점검·인터페이스 명세·브리핑 요약을 출력한다.
```python
"""데이터 갱신, 정합성 검증, 드리프트 점검, 자동 재학습(챔피언/챌린저), 7일 수송 소요 예측, 병목·우회 배분, 승인대기 분리까지 한 번에 실행하는 통합 파이프라인을 완성하고 운영 문서 완비율·자체 점검·인터페이스 명세·브리핑 요약을 출력한다."""
import importlib
import json
import sys
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import linprog
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.linear_model import Ridge
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 노선망·구간 통과량 생성기 (실제 도로·지명·부대 없음)
# 가상 출발지(SRC)에서 가상 목적지(DST)까지 3개 축선(A·B·C)과 연결로 2개, 구간 10개. 150일치 피크시간 통과량·지체 시간을 만든다.
# ---------------------------------------------------------------------------
SEGS = pd.DataFrame([
("A1", "SRC", "J1", 12, 130, "일반"), ("A2", "J1", "J2", 8, 70, "교량"), ("A3", "J2", "DST", 10, 140, "교차로"),
("B1", "SRC", "J3", 15, 100, "일반"), ("B2", "J3", "J2", 9, 80, "터널"), ("B3", "J3", "DST", 14, 95, "일반"),
("C1", "SRC", "J4", 18, 70, "일반"), ("C2", "J4", "DST", 20, 75, "교량"),
("X1", "J1", "J3", 7, 50, "교차로"), ("X2", "J4", "J3", 10, 60, "일반")],
columns=["seg", "u", "v", "km", "cap", "kind"]) # cap = 피크시간 통과 용량(대/시, 교육용 가정값)
KSPD = {"일반": 55, "교량": 40, "터널": 45, "교차로": 45} # 구간 유형별 자유 주행 속도(km/h, 가정값)
SEGS["t0"] = SEGS["km"] / SEGS["kind"].map(KSPD) * 60 # 자유 주행 시간(분)
UTIL = dict(A1=.78, A2=.85, A3=.66, B1=.70, B2=.72, B3=.64, C1=.55, C2=.58, X1=.45, X2=.40) # 평시 이용률
def bpr(t0, v, cap, a=0.15, b=4.0):
"""BPR 지체 함수(교통공학의 표준형): 통과량/용량 비가 커질수록 지체가 급증한다. (분)"""
return t0 * a * (v / cap) ** b
def make_data(seed=11, days=150):
"""구간 x 일자 패널. v=실제 피크 통과량, plan_v=수송계획 수량(긴급 불출 미반영), delay=실제 지체(분)."""
rng = np.random.default_rng(seed)
d_idx = np.arange(days)
lvl = np.array([1.0, 1.08, 1.04, 1.10, 1.02, 0.82, 0.78])[d_idx % 7] * (1 + 0.0012 * d_idx)
lvl = lvl * np.where((d_idx >= 100) & (d_idx < 110), 1.30, 1.0) # 수송량 폭증 기간(가상)
prev, rows = {s: 0.0 for s in SEGS["seg"]}, []
for d in d_idx:
rain = bool(rng.random() < 0.18)
for r in SEGS.itertuples():
maint, urgent = bool(rng.random() < 0.06), bool(rng.random() < 0.05) # 비계획 정비 / 긴급 불출
v = r.cap * UTIL[r.seg] * lvl[d] * (1.25 if urgent else 1.0) * rng.normal(1, 0.05)
cap_eff = r.cap * (0.65 if maint else 1.0) * (0.92 if rain else 1.0)
delay = max(r.t0 * 0.2 * (v / cap_eff) ** 3.5 + 0.25 * prev[r.seg] + rng.normal(0, 0.4), 0.0)
prev[r.seg] = delay
rows.append(dict(day=int(d), dow=int(d % 7), seg=r.seg, kind=r.kind, km=r.km, cap=r.cap, t0=r.t0,
plan_v=v / (1.25 if urgent else 1.0) * rng.normal(1, 0.04), v=v, maint=int(maint), urgent=int(urgent),
rain=int(rain), cap_eff=cap_eff, delay=delay))
df = pd.DataFrame(rows)
df["transit"] = df["t0"] * (1 + 0.08 * df["rain"]) + df["delay"]
return df
def mape(y, p):
y, p = np.asarray(y, float), np.asarray(p, float)
return float(np.mean(np.abs(y - p) / np.abs(y)) * 100)
def mae(y, p):
return float(np.mean(np.abs(np.asarray(y, float) - np.asarray(p, float))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 지체 예측용 파생 피처 생성(당일 오전 시점에 알 수 있는 정보만 사용: 계획 물량, 공지 정비, 기상 예보, 전일·전주 실적)
# ---------------------------------------------------------------------------
FEATS = ["plan_v", "cap", "cap_eff", "t0", "km", "maint", "rain", "util", "util_eff", "util_eff4", "bpr_base", "lag1_delay", "lag7_delay",
"roll7_v", "surge_ratio", "is_weekend", "dow_sin", "dow_cos", "k_교량", "k_터널", "k_교차로"]
def add_features(df):
d = df.sort_values(["seg", "day"]).copy()
g = d.groupby("seg")
d["lag1_delay"] = g["delay"].shift(1)
d["lag7_delay"] = g["delay"].shift(7)
d["roll7_v"] = g["v"].transform(lambda s: s.shift(1).rolling(7).mean())
d["util"] = d["plan_v"] / d["cap"]
d["util_eff"] = d["plan_v"] / d["cap_eff"]
d["util_eff4"] = d["util_eff"] ** 4
d["bpr_base"] = bpr(d["t0"], d["plan_v"], d["cap"])
d["surge_ratio"] = d["plan_v"] / d["roll7_v"]
d["is_weekend"] = (d["dow"] >= 5).astype(int)
d["dow_sin"], d["dow_cos"] = np.sin(2 * np.pi * d["dow"] / 7), np.cos(2 * np.pi * d["dow"] / 7)
for k in ("교량", "터널", "교차로"):
d["k_" + k] = (d["kind"] == k).astype(int)
return d.dropna().sort_values(["day", "seg"]).reset_index(drop=True)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# 노선망 흐름 도구: 최대유량(Edmonds-Karp), 증분 배정(혼잡 반영 최단시간 경로), 경로 열거
# ---------------------------------------------------------------------------
CAP = dict(zip(SEGS["seg"], SEGS["cap"]))
T0 = dict(zip(SEGS["seg"], SEGS["t0"]))
KM = dict(zip(SEGS["seg"], SEGS["km"]))
ARCS = {r.seg: (r.u, r.v) for r in SEGS.itertuples()}
SRC, DST = "SRC", "DST"
def seg_time(s, flow, cap=None, a=0.2, b=3.5):
"""구간 통과시간(분) = 자유주행 + 혼잡 지체. cap을 주면 용량을 대체(정비·기상 반영)."""
r = flow / (cap or CAP[s])
return T0[s] * (1 + a * r ** b + 6.0 * max(0.0, r - 1.0)) # 용량 초과분은 대기열로 급증
def max_flow(closed=(), capmul=None):
"""최대유량(대/시)과 최소 절단 구간(병목). capmul = {구간: 용량 배율}."""
cap = {s: (0 if s in closed else CAP[s] * (capmul or {}).get(s, 1.0)) for s in CAP}
res = {}
for s, (u, v) in ARCS.items():
res[(u, v)] = res.get((u, v), 0) + cap[s]
res.setdefault((v, u), 0)
flow = 0.0
while True:
prev, q = {SRC: None}, [SRC]
for u in q:
for (a, b), c in res.items():
if a == u and c > 1e-9 and b not in prev:
prev[b] = u; q.append(b)
if DST not in prev:
break
path, x = [], DST
while prev[x]:
path.append((prev[x], x)); x = prev[x]
f = min(res[e] for e in path)
for a, b in path:
res[(a, b)] -= f; res[(b, a)] += f
flow += f
reach = set(prev)
cut = [s for s, (u, v) in ARCS.items() if u in reach and v not in reach and cap[s] > 0]
return flow, cut
def all_paths():
"""SRC -> DST 단순 경로(구간 이름 튜플) 전체."""
out = []
def dfs(node, segs, seen):
if node == DST:
out.append(tuple(segs)); return
for s, (u, v) in ARCS.items():
if u == node and v not in seen:
dfs(v, segs + [s], seen | {v})
dfs(SRC, [], {SRC})
return out
PATHS = all_paths()
def assign(demand, closed=(), capmul=None, steps=40):
"""증분 배정: 수요를 steps조각으로 나눠, 전체 통과시간(대수 x 시간) 증가가 가장 작은 경로에 차례로 얹는다(시스템 최적 근사)."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
flow, pf = {s: 0.0 for s in CAP}, {}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for _ in range(steps):
chunk = demand / steps
best = min(ok, key=lambda p: sum((flow[s] + chunk) * seg_time(s, flow[s] + chunk, cap[s]) - flow[s] * seg_time(s, flow[s], cap[s]) for s in p))
for s in best:
flow[s] += chunk
pf[best] = pf.get(best, 0) + chunk
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
avg = sum(pf[p] * tt[p] for p in pf) / demand
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": avg, "util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
FIXED_SHARE = {("A1", "A2", "A3"): 0.32, ("B1", "B3"): 0.40, ("C1", "C2"): 0.28} # 과거 관행: 축선별 고정 배분(가상)
def fixed_assign(demand, closed=(), capmul=None):
"""과거 관행 배분: 축선별 고정 비율. 막힌 축선 물량은 열려 있는 축선에 비율대로 재분배."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
live = {p: w for p, w in FIXED_SHARE.items() if not any(s in closed for s in p)}
if not live:
return None
tot = sum(live.values())
flow = {s: 0.0 for s in CAP}
pf = {p: demand * w / tot for p, w in live.items()}
for p, f in pf.items():
for s in p:
flow[s] += f
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}}
def lp_assign(demand, closed=(), capmul=None):
"""선형계획(LP): 구간 용량 상한(이용률 rho) 아래에서 자유주행시간 합을 최소화하는 경로별 물량. 불가능하면 rho를 1.0까지 완화."""
cap = {s: CAP[s] * (capmul or {}).get(s, 1.0) for s in CAP}
ok = [p for p in PATHS if not any(s in closed for s in p)]
if not ok:
return None
for rho in (0.85, 0.95, 1.0):
r = linprog([sum(T0[s] for s in p) for p in ok], A_ub=[[1.0 if s in p else 0.0 for p in ok] for s in CAP],
b_ub=[rho * cap[s] for s in CAP], A_eq=[[1.0] * len(ok)], b_eq=[demand], bounds=(0, None), method="highs")
if r.status == 0:
break
else:
return assign(demand, closed, capmul) # 용량 초과 수요: 증분 배정으로 대체(과부하 표시)
flow = {s: sum(r.x[i] for i, p in enumerate(ok) if s in p) for s in CAP}
pf = {p: float(r.x[i]) for i, p in enumerate(ok) if r.x[i] > 1e-6}
tt = {p: sum(seg_time(s, flow[s], cap[s]) for s in p) for p in pf}
return {"flow": flow, "paths": pf, "path_min": tt, "avg_min": sum(pf[p] * tt[p] for p in pf) / demand,
"util": {s: flow[s] / cap[s] for s in cap if cap[s] > 0}, "rho": rho}
def cost_of(res, tv=1000, fuel_km=450):
"""시간당 비용(원, 교육용 가정): 대당 시간비용 tv원/분 + 유류비 fuel_km원/km."""
return sum(f * (res["path_min"][p] * tv + sum(KM[s] for s in p) * fuel_km) for p, f in res["paths"].items())
# ---------------------------------------------------------------------------
libs = ["numpy", "pandas", "scipy", "sklearn", "matplotlib"]
env_ok = sum(importlib.util.find_spec(m) is not None for m in libs)
print(f"[0] 환경 점검: Python {sys.version_info.major}.{sys.version_info.minor}, 필수 모듈 {env_ok}/{len(libs)}")
# ---------------- 1) 데이터 갱신·정합성 검증(미션 1~2 요약) ----------------
raw = make_data(days=180) # day 0~149 기존 + day 150~179 신규 입력분
d = add_features(raw)
chk = {"(구간,일자) 유일": not raw.duplicated(["seg", "day"]).any(), "결측 0": int(raw.isna().sum().sum()) == 0, "통과량 범위": bool(((raw.v >= 0) & (raw.v <= 3 * raw.cap)).all()),
"통과시간 >= 자유주행": bool((raw.transit >= raw.t0 - 1e-9).all())}
print(f"[1] 신규 30일 포함 {len(raw)}행 정합성 검증 {sum(chk.values())}/{len(chk)} 통과")
# ---------------- 2) 자동 재학습: 드리프트 점검 -> 챌린저 학습 -> 승격 판단 ----------------
def fit(tr):
return GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["delay"])
def evaluate(m, te):
p = te["t0"] * (1 + 0.08 * te["rain"]) + m.predict(te[FEATS])
return mae(te["transit"], p), mape(te["transit"], p)
def retrain_pipeline(old_end, new_end, drift_mape=3.0, min_gain=0.03):
old, new = d[d.day < old_end], d[(d.day >= old_end) & (d.day < new_end)]
champ = fit(old)
ch_mae, ch_mape = evaluate(champ, new)
drift = ch_mape > drift_mape
chall = fit(d[d.day < new_end - 10]) # 최근 10일은 검증용으로 남김
hold = d[(d.day >= new_end - 10) & (d.day < new_end)]
a, b = evaluate(champ, hold)[0], evaluate(chall, hold)[0]
promote = (a - b) / a >= min_gain
return {"champ_mape": ch_mape, "drift": drift, "champ_mae": a, "chall_mae": b, "promote": promote, "model": chall if promote else champ}
rt = retrain_pipeline(150, 180)
print(f"[2] 재학습 파이프라인: 기존 모델의 신규 30일 MAPE {rt['champ_mape']:.2f}% (드리프트 기준 3.0%: {'초과' if rt['drift'] else '이내'})")
print(f" 검증 10일 MAE 챔피언 {rt['champ_mae']:.3f}분 vs 챌린저 {rt['chall_mae']:.3f}분 -> {'챌린저 승격' if rt['promote'] else '기존 모델 유지'}(개선 3% 이상일 때만 승격)")
model = rt["model"]
# ---------------- 3) 7일 수송 소요 예측(Ridge 요일+추세 vs 전주 동요일 Baseline) ----------------
daily = raw[raw.seg.isin(["A1", "B1", "C1"])].groupby("day")["plan_v"].sum()
def design(days):
days = np.asarray(days)
X = np.zeros((len(days), 9))
X[np.arange(len(days)), days % 7] = 1
X[:, 7] = days / 100
X[:, 8] = ((days >= 100) & (days < 110)).astype(float) # 폭증 기간(알려진 사건) 더미
return X
tr_days = daily.index[daily.index < 173]
rg = Ridge(alpha=0.1, fit_intercept=False).fit(design(tr_days), daily[tr_days])
test_days = np.arange(173, 180)
f_ridge = rg.predict(design(test_days)); f_naive = daily[test_days - 7].values
print(f"\n[3] 수송 소요 7일 예측 검증(day 173~179): Ridge MAPE {mape(daily[test_days], f_ridge):.1f}% vs 전주 동요일 {mape(daily[test_days], f_naive):.1f}%")
rg = Ridge(alpha=0.1, fit_intercept=False).fit(design(daily.index), daily)
nxt = np.arange(180, 187)
dem = rg.predict(design(nxt))
# ---------------- 4) 소요 -> 병목·우회 배분 -> 승인대기 분리 ----------------
NOTICE = {183: {"A2": 0.65}} # 공지된 정비(가정): day 183 A2 교량 용량 65%
out = []
for day, D in zip(nxt, dem):
cm = NOTICE.get(int(day))
f, l = fixed_assign(D, (), cm), lp_assign(D, (), cm)
extra = l["avg_min"] - lp_assign(D)["avg_min"]
risky = max(l["util"].values()) > 0.98 or extra >= 3 or day in NOTICE
out.append({"day": int(day), "예측 소요(대/시)": D, "고정(분)": f["avg_min"], "우회 배정(분)": l["avg_min"], "절감률": 1 - cost_of(l) / cost_of(f),
"최대 이용률": max(l["util"].values()), "처리": "작전군수관 승인대기(2단계)" if risky else "추천(검토 후 집행)"})
pl = pd.DataFrame(out)
print("\n[4] 향후 7일 수송 소요·우회 배분 계획 [교육용가상자료]")
print(pl.round(2).to_string(index=False))
pend = pl[pl["처리"].str.contains("승인대기")]
print(f" 승인대기 {len(pend)}건(day {', '.join(map(str, pend['day']))}) - 자동 집행하지 않고 작전군수관 결재(미션 7)로 넘긴다.")
# ---------------- 5) 운영 문서 완비율·인터페이스 명세 ----------------
docs = {"데이터 사전·스키마(M1)": 29, "정제 규칙 10종(M2)": 10, "Baseline·EDA(M3)": 8, "병목 식별·우회(M4)": 4, "우회 소요·유류 엔진(M5)": 4, "대시보드·설명(M6)": 6,
"승인 절차·권한(M7)": 4, "스트레스 시험(M8)": 4, "AAR·ROI(M9)": 5, "운영·재학습 절차(M10)": 6, "장애 대응": 5, "인터페이스 가이드": 4}
done = sum(v > 0 for v in docs.values())
iface = {"input": {"daily_plan": ["day:int", "seg:str", "plan_v:float", "maint:int", "rain:int"], "format": "json-lines"},
"output": {"forecast": ["day", "demand_vph", "paths", "avg_min", "status"], "approval": "PENDING_ONLY"}, "auth": "role-based(가상)", "note": "상위 군수정보체계는 가상 인터페이스로만 정의"}
print(f"\n[5] 운영 문서 {len(docs)}개 항목 중 작성 완료 {done}개 ({done / len(docs):.0%}) / 인터페이스 명세(JSON):")
print(" ", json.dumps(iface, ensure_ascii=False))
# ---------------- 6) 자체 점검·브리핑 ----------------
te = d[d.day >= 150]
fin_mae, fin_mape = evaluate(model, te)
base_mape = mape(te.transit, te.t0 + te.bpr_base)
checks = [("환경 점검 통과", env_ok == len(libs)), ("데이터 정합성 전 항목 통과", all(chk.values())), ("재학습 파이프라인 정상 실행(챔피언/챌린저 판정 출력)", rt["champ_mae"] > 0),
("통과시간 예측 MAPE 25% 이하", fin_mape <= 25), ("기준 BPR 대비 MAE 20% 이상 개선", fin_mae <= 0.8 * mae(te.transit, te.t0 + te.bpr_base)),
("7일 소요 예측이 전주 동요일 Baseline보다 정확", mape(daily[test_days], f_ridge) < mape(daily[test_days], f_naive)),
("위험·정비 우회안 자동 집행 0건(승인대기 분리)", len(pend) >= 1 and (pl.loc[pl.day == 183, "처리"].str.contains("승인대기")).all()), ("운영 문서 100% 완비", done == len(docs))]
print("\n[6] 자체 점검")
for n, ok in checks:
print(f" [{'PASS' if ok else 'FAIL'}] {n}")
print("\n[7] 브리핑 요약")
print(f" - 신규 30일 데이터로 재학습 판단을 자동 수행했다(드리프트 {'감지' if rt['drift'] else '미감지'}, {'챌린저 승격' if rt['promote'] else '기존 모델 유지'}).")
print(f" - 통과시간 MAPE {base_mape:.1f}%(BPR) -> {fin_mape:.1f}%, 향후 7일 소요 예측 MAPE {mape(daily[test_days], f_ridge):.1f}%.")
print(f" - 향후 7일 중 우회 최적화 평균 절감 {pl['절감률'].mean():.1%}, 정비 공지일은 승인대기로 분리했다. 실제 우회 하달은 작전군수관 결재와 지휘계통 결심에 따른다.")
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
ax[0].plot(daily.index, daily.values, label="actual"); ax[0].plot(nxt, dem, "r--o", label="forecast"); ax[0].legend(); ax[0].set_title("Daily transport demand and 7-day forecast")
ax[1].bar(pl["day"] - .2, pl["고정(분)"], .4, label="fixed"); ax[1].bar(pl["day"] + .2, pl["우회 배정(분)"], .4, label="detour assignment"); ax[1].legend(); ax[1].set_title("Avg transit (min), next 7 days")
plt.suptitle("MOV01 integrated pipeline (synthetic)"); plt.tight_layout(); plt.show()
```
실행 결과 예시
```text
[0] 환경 점검: Python 3.11, 필수 모듈 5/5
[1] 신규 30일 포함 1800행 정합성 검증 4/4 통과
[2] 재학습 파이프라인: 기존 모델의 신규 30일 MAPE 2.93% (드리프트 기준 3.0%: 이내)
검증 10일 MAE 챔피언 0.534분 vs 챌린저 0.535분 -> 기존 모델 유지(개선 3% 이상일 때만 승격)
[3] 수송 소요 7일 예측 검증(day 173~179): Ridge MAPE 3.5% vs 전주 동요일 4.7%
[4] 향후 7일 수송 소요·우회 배분 계획 [교육용가상자료]
day 예측 소요(대/시) 고정(분) 우회 배정(분) 절감률 최대 이용률 처리
180 216.46 42.89 42.43 0.00 0.85 추천(검토 후 집행)
181 204.40 42.15 41.95 0.00 0.85 추천(검토 후 집행)
182 258.65 56.35 44.70 0.16 0.95 추천(검토 후 집행)
183 276.73 91.80 51.32 0.37 1.11 작전군수관 승인대기(2단계)
184 266.37 61.40 45.29 0.21 0.95 추천(검토 후 집행)
185 275.57 68.96 45.87 0.27 1.00 작전군수관 승인대기(2단계)
186 259.89 57.16 44.79 0.17 0.95 추천(검토 후 집행)
승인대기 2건(day 183, 185) - 자동 집행하지 않고 작전군수관 결재(미션 7)로 넘긴다.
[5] 운영 문서 12개 항목 중 작성 완료 12개 (100%) / 인터페이스 명세(JSON):
{"input": {"daily_plan": ["day:int", "seg:str", "plan_v:float", "maint:int", "rain:int"], "format": "json-lines"}, "output": {"forecast": ["day", "demand_vph", "paths", "avg_min", "status"], "approval": "PENDING_ONLY"}, "auth": "role-based(가상)", "note": "상위 군수정보체계는 가상 인터페이스로만 정의"}
[6] 자체 점검
[PASS] 환경 점검 통과
[PASS] 데이터 정합성 전 항목 통과
[PASS] 재학습 파이프라인 정상 실행(챔피언/챌린저 판정 출력)
[PASS] 통과시간 예측 MAPE 25% 이하
[PASS] 기준 BPR 대비 MAE 20% 이상 개선
[PASS] 7일 소요 예측이 전주 동요일 Baseline보다 정확
[PASS] 위험·정비 우회안 자동 집행 0건(승인대기 분리)
[PASS] 운영 문서 100% 완비
[7] 브리핑 요약
- 신규 30일 데이터로 재학습 판단을 자동 수행했다(드리프트 미감지, 기존 모델 유지).
- 통과시간 MAPE 9.6%(BPR) -> 2.9%, 향후 7일 소요 예측 MAPE 3.5%.
- 향후 7일 중 우회 최적화 평균 절감 16.9%, 정비 공지일은 승인대기로 분리했다. 실제 우회 하달은 작전군수관 결재와 지휘계통 결심에 따른다.
```
Reference in new issue View Git Blame Copy Permalink