243 KiB
243 KiB
🎯 [MOV06] 납품수요 예측·물류흐름 추적관리
Google Colab 실습 환경 (수송·물류·창고 / Level 3)
📌 1. 프로젝트 개요 및 배경
| 항목 | 내용 |
|---|---|
| 과정 분류 | 수송·물류·창고 (Level 3) |
| 학습 소요 시간 | 10개 미션 (5일 / 총 28.5시간) |
| 환경 사양 | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |
📖 프로젝트 목표
납품물류에서는 수령부대의 납품 소요를 예측하는 일과 수송선단이 지금 어디에 있고 언제 도착하는지 추적하는 일이 따로 이루어지면 지연을 뒤늦게 알게 된다. 본 프로젝트는 납품수요 예측과 물류흐름 추적을 결합해 end-to-end 물류 가시성을 익히도록, 교육용 가상 노선에서 선단의 GPS 위치·화물 센서·통신 상태 텔레메트리를 코드로 만들어 도착 예정 시간(ETA)과 지연 원인을 추적한다. 학습자는 텔레메트리 데이터셋 정의, GPS 이상값 정제, 현재 속도 기반 ETA 기준선, 도로·기상·정체를 반영한 ETA 모델, 이탈·불시 정차 경보, 관제 대시보드와 지연 요인 설명, 현장 확인·대체 수송 승인(HITL), 100개 선단 스트림 부하 시험, AAR 기반 가중치 재튜닝과 ROI, 재학습 파이프라인 통합을 차례로 수행한다. 최종 목표는 선단 이동 상태를 추적해 도착 시각과 지연 원인을 설명하고 7일 납품 소요까지 예측하되, 대체 수송 지시는 반드시 사람이 결정하는 물류 추적 도구를 완성하는 것이다.
⚠️ [교육용 가상 데이터셋 원칙]
- 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 코드 내 난수 기반 가상 데이터로 생성됩니다.
- 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.
In [ ]:
# ============================================================================
# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트
# ============================================================================
!pip install -q numpy pandas matplotlib scipy scikit-learn
import os
import sys
import time
import math
import json
import re
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')
# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)
plt.rcParams['font.sans-serif'] = 'DejaVu Sans'
plt.rcParams['axes.unicode_minus'] = False
print(f"실행 환경 준비 완료 (Python {sys.version.split()[0]})")
In [ ]:
"""군수 수송선단의 GPS 위치·화물 센서·통신 상태 텔레메트리 데이터셋을 정의하고, 허용값 검증기와 주입 오류 시험, 실제 생성 데이터와의 대조, 납품 물류 흐름 추적 업무 프로세스 맵(20개 이상 피처 목록 포함)을 만든다."""
import copy
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 1] 군수 수송선단 GPS 위치, 화물 센서, 통신 상태 텔레메트리 데이터셋 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 1] 군수 수송선단 GPS 위치, 화물 센서, 통신 상태 텔레메트리 데이터셋 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
# 1) 데이터 사전: (필드, 구분, 형, 허용 범위/값, 설명)
DICT = [("trip", "식별", "int", "0~", "수송선단 이동 번호(날짜 순)"), ("t", "GPS", "int", "5~720", "출발 후 경과 시간(분)"), ("clock", "GPS", "int", "0~1439", "보고 시각(0시부터 분)"),
("x", "GPS", "float", "-10~160", "가상 좌표 x(km)"), ("y", "GPS", "float", "-10~90", "가상 좌표 y(km)"), ("speed", "GPS", "float", "0~120", "보고 속도(km/h)"),
("heading", "GPS", "float", "0~360", "진행 방향(도)"), ("rssi", "통신", "float", "-110~-40", "통신 수신 강도(dBm)"), ("cargo_temp", "화물센서", "float", "-5~40", "화물칸 온도(C)"),
("shock_g", "화물센서", "float", "0~5", "충격 가속도(g)"), ("rain", "맥락", "int", "0|1", "우천 여부"), ("size", "맥락", "int", "3~12", "선단 차량 수"),
("cargo", "맥락", "str", "일반|냉장|민감", "화물 구분"), ("s_true", "라벨", "float", "0~LEN", "실제 진행거리(모의 정답)"), ("off_true", "라벨", "float", "0~", "실제 노선 이탈거리(모의 정답)"),
("stopped_true", "라벨", "int", "0|1", "실제 정차 여부(모의 정답)"), ("stop_cause", "라벨", "str", "maint|urgent|''", "정차 원인(비계획 정비/긴급 불출 대기)"), ("eta_true", "라벨", "int", "0~", "실제 도착까지 남은 시간(분)"),
("s_obs", "파생", "float", "0~LEN", "GPS를 노선에 투영한 진행거리"), ("off_km", "파생", "float", "0~", "노선과의 수직 이탈거리"), ("remaining_km", "파생", "float", "0~LEN", "남은 거리"),
("progress", "파생", "float", "0~1", "진행률"), ("speed_ma3", "파생", "float", "0~", "최근 3회 평균 속도"), ("speed_std3", "파생", "float", "0~", "최근 3회 속도 표준편차"),
("is_stop", "파생", "int", "0|1", "정차 판정(3km/h 미만)"), ("stop_n", "파생", "int", "0~", "누적 정차 횟수"), ("stop_min", "파생", "int", "0~", "누적 정차 시간(분)"),
("since_stop", "파생", "int", "0~", "마지막 정차 이후 경과(분)"), ("hour_sin", "파생", "float", "-1~1", "시각 순환(사인)"), ("hour_cos", "파생", "float", "-1~1", "시각 순환(코사인)"),
("cong", "파생", "float", "0.5~1", "시간대 정체 계수"), ("k_냉장", "파생", "int", "0|1", "냉장 화물 더미"), ("k_민감", "파생", "int", "0|1", "민감 화물 더미"),
("rem_unpaved_km", "파생", "float", "0~", "남은 비포장 거리"), ("rem_threat_km", "파생", "float", "0~", "남은 위협(정체) 구역 거리"), ("rem_bridge_km", "파생", "float", "0~", "남은 교량·터널 거리"),
("base_eta", "파생", "float", "0~600", "현재 속도 기반 단순 ETA(분)")]
dd = pd.DataFrame(DICT, columns=["필드", "구분", "형", "허용값", "설명"])
cnt = dd["구분"].value_counts().to_dict()
print(f"[1] 데이터 사전 {len(dd)}개 필드: {cnt}")
# 2) 실제 생성 데이터와 대조
raw = make_telemetry(n_trips=20)
feat = add_features(raw)
extra_cols = sorted(set(feat.columns) - set(dd["필드"]))
missing = sorted(set(dd["필드"]) - set(feat.columns))
fe = dd[~dd["구분"].isin(["식별", "라벨"])]
print(f"[2] 데이터 사전 대조({len(raw)}행): 사전 누락 컬럼 {extra_cols or '없음'}, 데이터에 없는 필드 {missing or '없음'}")
print(f" 예측·추적 피처 후보 {len(fe)}개(목표 20개 이상: {'달성' if len(fe) >= 20 else '미달'}), 라벨(모의 정답) {int((dd['구분'] == '라벨').sum())}개는 실제 운용에서는 존재하지 않는다")
# 3) 허용값 검증기와 오류 주입
RULES = {"x": (-10, 160), "y": (-10, 90), "speed": (0, 120), "heading": (0, 360), "rssi": (-110, -40), "cargo_temp": (-5, 40), "shock_g": (0, 5), "t": (5, 720), "size": (3, 12)}
def validate(rec):
errs = []
for k, (lo, hi) in RULES.items():
v = rec.get(k)
if v is None or (isinstance(v, float) and np.isnan(v)):
errs.append(f"{k}: 결측")
elif not lo <= v <= hi:
errs.append(f"{k}: 범위 밖({v:.1f})")
if rec.get("cargo") not in ("일반", "냉장", "민감"):
errs.append("cargo: 허용되지 않은 값")
return errs
recs = raw.drop(columns=["stop_cause"]).to_dict("records")
n_err = sum(1 for r in recs if validate(r))
print(f"[3] 정상 샘플 {len(recs)}건 검증 오류 {n_err}건")
bad = [("GPS 좌표 sentinel", {**recs[0], "x": -999.0, "y": -999.0}), ("속도 단위 오류", {**recs[1], "speed": 1800.0}), ("방향 범위 초과", {**recs[2], "heading": 400.0}),
("센서 sentinel", {**recs[3], "cargo_temp": -999.0}), ("속도 결측", {**recs[4], "speed": float("nan")}), ("화물 구분 오기", {**recs[5], "cargo": "기타"})]
caught = 0
for name, r in bad:
e = validate(copy.deepcopy(r))
caught += bool(e)
print(f" - {name}: {'; '.join(e)}")
print(f" 주입 오류 {len(bad)}종 중 {caught}종 검출")
# 4) 납품 물류 흐름 추적 업무 프로세스 맵
proc = [("납품 수요 접수", "수송계획장교", "수령부대 요청 서식", "일일 납품 소요", 30), ("선단 편성·출발 보고", "수송통제소", "자산·화물 목록", "trip, size, cargo", 25),
("GPS·센서 텔레메트리 수신", "관제 담당", "5분 간격 보고", "x, y, speed, rssi", 5), ("데이터 정제·보정", "분석 담당", "원시 텔레메트리", "정제 데이터", 15),
("ETA 산출", "분석 담당", "정제 데이터, 노선 구성", "ETA(분)", 10), ("이탈·정차 감지", "관제 담당", "위치·속도", "이상 경보", 5),
("지연 경보 확인", "현장확인자", "경보, ETA 변동", "확인 결과", 30), ("대체 수송 지시", "수송통제관", "확인 결과", "지시(승인)", 20),
("도착 확인·실적 대조", "수령부대", "도착 보고", "ETA 오차 기록", 25)]
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프로세스 맵 작성 완료, 피처 목록 20개 이상 ->", "달성" if len(fe) >= 20 and len(pm) >= 5 else "미달")
print("모든 노선·좌표·수치는 교육용 가상 자료이며 실제 선단·부대·도로와 무관하다.")
In [ ]:
"""수송선단 텔레메트리의 비정상 GPS 좌표·속도·방향을 보정하는 전처리 모듈: 오염 기록에 정제 규칙 10종을 적용하고, 위치 차분으로 속도·방향을 재산출하며, 정차(비계획 정비·긴급 불출 대기) 기록은 삭제하지 않고 보존한다."""
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 2] 수송 데이터 정제, 비정상 GPS 좌표 보정 및 속도/이동 방향 전처리 모듈 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 2] 수송 데이터 정제, 비정상 GPS 좌표 보정 및 속도/이동 방향 전처리 모듈 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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_telemetry(n_trips=30)
rng = np.random.default_rng(3)
dirty = clean.drop(columns=["stop_cause"]).copy()
n = len(dirty)
ix = rng.choice(n, 90, replace=False)
dirty.loc[ix[:8], ["x", "y"]] = -999.0 # 좌표 sentinel
dirty.loc[ix[8:16], "x"] += rng.choice([-45.0, 45.0], 8) # GPS 점프(순간 이동)
dirty.loc[ix[16:22], "speed"] = -dirty.loc[ix[16:22], "speed"] - 1 # 음수 속도
dirty.loc[ix[22:28], "speed"] = dirty.loc[ix[22:28], "speed"] * 36 + 100 # 단위 오류(과대)
dirty.loc[ix[28:36], "speed"] = np.nan
dirty.loc[ix[36:44], "heading"] = rng.choice([-30.0, 400.0, np.nan, 725.0], 8)
dirty.loc[ix[44:52], "rssi"] = -999.0
dirty.loc[ix[52:58], "cargo_temp"] = -999.0
dirty = dirty.drop(index=ix[58:70]).reset_index(drop=True) # 패킷 누락 12건
dirty = pd.concat([dirty, dirty.sample(10, random_state=1)], ignore_index=True) # 중복 전송 10건
dirty = dirty.sample(frac=1, random_state=2).reset_index(drop=True) # 도착 순서 뒤섞임(순서 오류)
print(f"[0] 오염 데이터 {len(dirty)}행(원본 {n}행): 좌표 sentinel 8, GPS 점프 8, 속도 음수 6·단위 오류 6·결측 8, 방향 이상 8, 센서 sentinel 14, 누락 12, 중복 10, 순서 뒤섞임")
log, w = [], dirty.copy()
def rule(name, k):
log.append((len(log) + 1, name, int(k)))
# R1 정렬 + (trip, t) 중복 제거
n0 = len(w); w = w.sort_values(["trip", "t"]).drop_duplicates(["trip", "t"]).reset_index(drop=True); rule("시간순 정렬·(trip,t) 중복 제거", n0 - len(w))
# R2 센서 sentinel -> 결측
bad = (w["rssi"] <= -900) | (w["cargo_temp"] <= -900); w.loc[w["rssi"] <= -900, "rssi"] = np.nan; w.loc[w["cargo_temp"] <= -900, "cargo_temp"] = np.nan; rule("센서 sentinel(-999) -> 결측", bad.sum())
# R3 좌표 범위 밖 -> 결측
bad = ~(w["x"].between(-10, 160) & w["y"].between(-10, 90)); w.loc[bad, ["x", "y"]] = np.nan; rule("좌표 범위 밖(sentinel 등) -> 결측", bad.sum())
# R4 GPS 점프: 앞뒤 5개 중앙값에서 8km 이상 벗어난 좌표(노선 이탈은 최대 4km이므로 보존)
med_x = w.groupby("trip")["x"].transform(lambda s: s.rolling(5, center=True, min_periods=2).median())
med_y = w.groupby("trip")["y"].transform(lambda s: s.rolling(5, center=True, min_periods=2).median())
jump = (np.hypot(w["x"] - med_x, w["y"] - med_y) > 8) & w["x"].notna(); w.loc[jump, ["x", "y"]] = np.nan; rule("GPS 점프(이웃 중앙값에서 8km 초과) -> 결측", jump.sum())
# R5 좌표 결측 선형 보간(시간 기준)
miss = w["x"].isna(); w[["x", "y"]] = w.groupby("trip")[["x", "y"]].transform(lambda s: s.interpolate(limit_direction="both")); rule("좌표 결측 -> 시간 선형 보간", miss.sum())
# R6 위치 차분으로 속도·방향 산출
dt = w.groupby("trip")["t"].diff(); dx, dy = w.groupby("trip")["x"].diff(), w.groupby("trip")["y"].diff()
v_pos = np.hypot(dx, dy) / (dt / 60); h_pos = np.degrees(np.arctan2(dx, dy)) % 360
rule("보고 속도와 위치 차분 속도 30km/h 초과 불일치 표시", ((w["speed"] - v_pos).abs() > 30).sum())
# R7 속도 음수·단위 오류(120 초과) -> 결측
bad = (w["speed"] < 0) | (w["speed"] > 120); w.loc[bad, "speed"] = np.nan; rule("속도 음수·120km/h 초과 -> 결측", bad.sum())
# R8 속도 결측 -> 위치 차분 속도(상한 120), 그래도 없으면 직전값
miss = w["speed"].isna(); w.loc[miss, "speed"] = v_pos[miss].clip(0, 120); w["speed"] = w.groupby("trip")["speed"].ffill().bfill(); rule("속도 결측 -> 위치 차분 속도 대체", miss.sum())
# R9 방향 정규화(0~360), 범위 밖·결측은 위치 차분 방향
bad = ~w["heading"].between(0, 360, inclusive="left"); w.loc[bad, "heading"] = h_pos[bad]; w["heading"] = w.groupby("trip")["heading"].ffill().bfill() % 360; rule("방향 범위 밖·결측 -> 위치 차분 방향", bad.sum())
# R10 정차 기록 보존 표시 + 패킷 누락 구간 표시(삭제 없음)
w["is_stop"] = (w["speed"] < 3).astype(int); w["gap"] = (w.groupby("trip")["t"].diff() > 5).astype(int)
w["rssi"] = w.groupby("trip")["rssi"].ffill().bfill(); w["cargo_temp"] = w.groupby("trip")["cargo_temp"].ffill().bfill()
rule("정차 행 보존 표시·누락 구간 표시(삭제 없음)", w["is_stop"].sum() + w["gap"].sum())
print("\n[1] 정제 규칙 적용 로그")
print(pd.DataFrame(log, columns=["규칙", "내용", "처리 건수"]).to_string(index=False))
print(f" 정제 규칙 {len(log)}종 적용(목표 10종: {'달성' if len(log) >= 10 else '미달'})")
checks = {"(trip,t) 유일": not w.duplicated(["trip", "t"]).any(), "시간 단조 증가": bool((w.groupby("trip")["t"].diff().dropna() > 0).all()),
"좌표 범위": bool(w["x"].between(-10, 160).all() and w["y"].between(-10, 90).all()), "속도 0~120": bool(w["speed"].between(0, 120).all()),
"방향 0~360": bool(w["heading"].between(0, 360, inclusive="left").all()), "통신 강도 범위": bool(w["rssi"].between(-110, -40).all()),
"화물 온도 범위": bool(w["cargo_temp"].between(-5, 40).all()), "필수 컬럼 결측 0": int(w[["x", "y", "speed", "heading", "rssi", "cargo_temp"]].isna().sum().sum()) == 0}
print("\n[2] 정합성 검증")
for k, v in checks.items():
print(f" [{'PASS' if v else 'FAIL'}] {k}")
print(f" 통과율 {sum(checks.values()) / len(checks):.0%} (목표 100%)")
# 보정 효과: 진행거리 오차(투영 진행거리 vs 실제 진행거리), 정차 보존
s_d, _ = project(dirty["x"].where(dirty["x"] > -900, np.nan).fillna(0).values, dirty["y"].where(dirty["y"] > -900, np.nan).fillna(0).values)
s_w, _ = project(w["x"].values, w["y"].values)
truth = clean.set_index(["trip", "t"])["s_true"]
tru_d = truth.reindex(pd.MultiIndex.from_frame(dirty[["trip", "t"]])).values
tru_w = truth.reindex(pd.MultiIndex.from_frame(w[["trip", "t"]])).values
print(f"\n[3] 진행거리 오차(MAE): 오염 {np.nanmean(np.abs(s_d - tru_d)):.2f}km -> 정제 후 {np.nanmean(np.abs(s_w - tru_w)):.2f}km")
stop_true = int(clean["stopped_true"].sum()); stop_w = int(w["is_stop"].sum())
print(f" 정차 기록 보존: 실제 정차 {stop_true}행(비계획 정비 {int((clean.stop_cause == 'maint').sum())}, 긴급 불출 대기 {int((clean.stop_cause == 'urgent').sum())}) / 정제 후 정차 행 {stop_w}행(차이 {stop_true - stop_w}행은 전송 누락분) -> 정차를 이상치로 삭제하지 않음")
print("[4] 전후 기술통계(speed)")
st = pd.DataFrame({"오염": dirty["speed"].describe(), "정제 후": w["speed"].describe(), "원본": clean["speed"].describe()}).loc[["count", "mean", "min", "max"]].round(1)
print(st.to_string())
In [ ]:
"""현재 속도 기반 단순 ETA(남은 거리 / 현재 속도) Baseline을 만들어 오차를 측정하고, 상황별 오차 비교·EDA 차트 8종·상관 피처·파레토 분석으로 Baseline의 한계와 핵심 지연 요인을 찾는다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import stats
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 3] 현재 속도 기반 단순 도착 예정 시간(ETA) 계산 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 3] 현재 속도 기반 단순 도착 예정 시간(ETA) 계산 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(); d = add_features(df)
d = d[d["eta_true"] >= 15].copy() # 도착 15분 이내는 상대오차가 과대해져 평가에서 제외
d["err"] = d["base_eta"] - d["eta_true"] # 양수 = 늦게 도착한다고 과대 예측
print(f"[1] Baseline ETA(남은 거리/현재 속도) 평가 {len(d)}행: MAPE {mape(d.eta_true, d.base_eta):.1f}%, MAE {mae(d.eta_true, d.base_eta):.1f}분, 평균 오차 {d.err.mean():+.1f}분")
cond = {"정차 중(속도 3 미만)": d.is_stop == 1, "직전 정차 이력 있음": d.stop_n > 0, "우천": d.rain == 1, "비포장 남은 거리 40km 이상": d.rem_unpaved_km >= 40,
"위협(정체) 구역 앞": d.rem_threat_km > 0, "노선 이탈 1km 이상": d.off_km >= 1, "그 외 평상": (d.is_stop == 0) & (d.stop_n == 0) & (d.rain == 0) & (d.off_km < 1)}
tab = pd.DataFrame([(k, int(m.sum()), mae(d.eta_true[m], d.base_eta[m]), d.err[m].mean()) for k, m in cond.items()], columns=["조건", "건수", "MAE(분)", "평균 오차(분)"]).round(1)
print("[2] 조건별 Baseline 오차(양수 = 과대 예측)")
print(tab.to_string(index=False))
print(" -> 정차·직전 정차 이력·비포장·위협 구역이 있으면 현재 속도만으로는 남은 시간을 크게 잘못 본다(한계).")
rows = []
for f in FEATS:
if f == "base_eta":
continue
r, p = stats.spearmanr(d[f], d["err"].abs())
rows.append((f, r, p))
cor = pd.DataFrame(rows, columns=["피처", "상관", "p"]).assign(a=lambda x: x["상관"].abs()).sort_values("a", ascending=False)
sig = cor[(cor.a >= 0.2) & (cor.p < 0.05)]
print(f"\n[3] Baseline 절대오차와 유의한 상관(|r|>=0.2, p<0.05) 피처 {len(sig)}개 (목표 5개 이상: {'달성' if len(sig) >= 5 else '미달'})")
print(sig.head(6)[["피처", "상관", "p"]].round(3).to_string(index=False))
print(" (남은 거리·진행률·남은 구간 구성은 '멀수록 오차가 큼'을 반영하므로 해석 때 거리 효과와 분리해서 본다)")
a, b = d.loc[d.stop_n > 0, "err"], d.loc[d.stop_n == 0, "err"]
t, p = stats.ttest_ind(a, b, equal_var=False)
print(f" 가설검정: 정차 이력 있음 vs 없음 오차 평균 {a.mean():+.1f}분 vs {b.mean():+.1f}분, t={t:.1f}, p={p:.2g}")
trip_err = d.groupby("trip")["err"].apply(lambda s: s.abs().sum()).sort_values(ascending=False)
top = int(np.ceil(len(trip_err) * 0.2))
print(f"\n[4] 파레토: 오차 상위 20% 이동({top}건)이 전체 ETA 오차의 {trip_err.iloc[:top].sum() / trip_err.sum():.0%}를 차지 -> 지연 이동 집중 관리 대상")
fig, ax = plt.subplots(2, 4, figsize=(18, 8))
for tid in (3, 17, 30):
q = df[df.trip == tid]; ax[0, 0].plot(q.t, q.s_true, label=f"trip {tid}")
ax[0, 0].set_title("1 Progress by trip (km)"); ax[0, 0].legend()
ax[0, 1].hist(df["speed"], bins=40); ax[0, 1].set_title("2 Reported speed distribution")
ax[0, 2].scatter(d["eta_true"], d["base_eta"], s=4, alpha=.3); ax[0, 2].plot([0, 400], [0, 400], "r--"); ax[0, 2].set_ylim(0, 650); ax[0, 2].set_title("3 Baseline ETA vs actual")
ax[0, 3].hist(d["err"].clip(-200, 300), bins=40); ax[0, 3].set_title("4 Baseline error (min)")
ax[1, 0].boxplot([d.loc[d.rain == 0, "err"].clip(-200, 300), d.loc[d.rain == 1, "err"].clip(-200, 300)]); ax[1, 0].set_xticklabels(["dry", "rain"]); ax[1, 0].set_title("5 Rain effect on error")
by_seg = df.assign(seg=[seg_of(s) for s in df.s_true]).groupby("seg")["speed"].mean(); ax[1, 1].bar([f"R{i + 1}" for i in by_seg.index], by_seg.values); ax[1, 1].set_title("6 Mean speed by route segment")
hs = df.groupby(df["clock"] // 60)["speed"].mean(); ax[1, 2].plot(hs.index, hs.values, "o-"); ax[1, 2].set_title("7 Mean speed by hour of day")
ax[1, 3].bar(range(len(trip_err)), (trip_err / trip_err.sum()).values[:], width=1.0); ax[1, 3].set_title("8 Pareto: error by trip (sorted)")
plt.suptitle("MOV06 EDA (synthetic telemetry)"); plt.tight_layout(); plt.show()
print("\n[5] EDA 차트 8종 생성 완료(목표 8종 이상: 달성)")
print("핵심 발견: 정차와 비포장·위협 구역이 남은 시간을 늘리지만 현재 속도에는 반영되지 않는다. 이는 M4 피처 설계(남은 구간 구성, 정차 이력)의 근거가 된다.")
In [ ]:
"""도로 상태(비포장·교량터널 잔여 거리), 기상(우천), 정체(시간대), 위협 구역 잔여 거리와 정차 이력을 반영한 머신러닝 정밀 ETA 모델을 만들고, 시계열 교차검증(이동 순서)으로 MAPE를 측정하며 변수군 제거 실험으로 각 변수군의 기여를 확인한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.inspection import permutation_importance
from sklearn.linear_model import Ridge
from sklearn.model_selection import TimeSeriesSplit
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 4] 도로 상태, 기상, 적 위협 정체 변수를 반영한 머신러닝 기반 정밀 ETA 예측 모델 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 4] 도로 상태, 기상, 적 위협 정체 변수를 반영한 머신러닝 기반 정밀 ETA 예측 모델 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(); d = add_features(df)
print(f"[1] 파생 피처 {len(FEATS) - 5}개 포함 총 {len(FEATS)}개 사용(남은 거리·구성, 정차 이력, 시간대 순환, 속도 이동평균·표준편차, 이탈거리 등). 평가는 도착 15분 이상 남은 관측만 사용")
trips = np.sort(d["trip"].unique())
tscv = TimeSeriesSplit(n_splits=4)
rows = []
for k, (a, b) in enumerate(tscv.split(trips)):
tr, te = d[d.trip.isin(trips[a])], d[d.trip.isin(trips[b])]
te = te[te.eta_true >= 15]
gbr = HistGradientBoostingRegressor(max_iter=150, max_depth=4, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["eta_true"])
rdg = Ridge(alpha=1.0).fit(tr[FEATS], tr["eta_true"])
rows.append((k + 1, len(trips[a]), len(trips[b]), mape(te.eta_true, te.base_eta), mape(te.eta_true, rdg.predict(te[FEATS])), mape(te.eta_true, gbr.predict(te[FEATS]))))
cv = pd.DataFrame(rows, columns=["fold", "학습 이동", "검증 이동", "단순 ETA MAPE(%)", "Ridge MAPE(%)", "GBR MAPE(%)"]).round(2)
print("[2] 시계열 교차검증(이동 순서대로 학습 구간이 늘어남, 미래 정보 누수 없음)")
print(cv.to_string(index=False))
mm = cv["GBR MAPE(%)"].mean()
print(f" GBR 평균 MAPE {mm:.2f}% (목표 25% 이하: {'달성' if mm <= 25 else '미달'}), 단순 ETA 평균 {cv['단순 ETA MAPE(%)'].mean():.1f}%")
# 변수군 제거 실험(학습 160 이동 / 검증 40 이동): 각 변수군을 빼면 MAPE가 얼마나 나빠지는가
tr, te = d[d.trip < 160], d[(d.trip >= 160) & (d.eta_true >= 15)]
groups = {"도로 상태(비포장·교량터널 잔여)": ["rem_unpaved_km", "rem_bridge_km"], "기상(우천)": ["rain"], "정체(시간대)": ["hour_sin", "hour_cos", "cong"],
"위협·정체 구역 잔여 거리": ["rem_threat_km"], "정차 이력": ["stop_n", "stop_min", "since_stop"], "이탈·통신": ["off_km", "rssi"]}
def fit_eval(cols):
m = HistGradientBoostingRegressor(max_iter=150, max_depth=4, learning_rate=0.08, random_state=0).fit(tr[cols], tr["eta_true"])
return mape(te.eta_true, m.predict(te[cols])), m
full, m_full = fit_eval(FEATS)
ab = [(g, fit_eval([f for f in FEATS if f not in cols])[0]) for g, cols in groups.items()]
abt = pd.DataFrame(ab, columns=["제거한 변수군", "MAPE(%)"]).assign(증가=lambda x: x["MAPE(%)"] - full).sort_values("증가", ascending=False).round(2)
print(f"\n[3] 변수군 제거 실험(전체 사용 MAPE {full:.2f}%)")
print(abt.to_string(index=False))
pi = permutation_importance(m_full, te[FEATS], te["eta_true"], n_repeats=3, random_state=0, scoring="neg_mean_absolute_error")
imp = pd.Series(pi.importances_mean, index=FEATS).sort_values(ascending=False)
imp = imp / imp.sum()
print(" 순열 중요도 상위 5:", ", ".join(f"{k} {v:.2f}" for k, v in imp.head(5).items()), f"/ 1% 미만 {int((imp < 0.01).sum())}개(차원 축소 후보)")
base_all = mape(te.eta_true, te.base_eta)
print(f"\n[4] 검증 40개 이동: 단순 ETA MAPE {base_all:.1f}% -> GBR {full:.1f}%, MAE {mae(te.eta_true, te.base_eta):.1f}분 -> {mae(te.eta_true, m_full.predict(te[FEATS])):.1f}분")
print(" 해석: 노선이 하나뿐이라 남은 비포장·위협·교량터널 거리는 진행률로 대체되어 단독 제거 효과가 작다. 노선이 여러 개면 이 변수들이 필수다. 우천과 시간대 정체는 진행률로 대체되지 않아 제거 시 오차가 커진다.")
fig, ax = plt.subplots(1, 3, figsize=(16, 4.5))
ax[0].plot(cv["fold"], cv["단순 ETA MAPE(%)"], "o-", label="simple"); ax[0].plot(cv["fold"], cv["GBR MAPE(%)"], "s-", label="GBR"); ax[0].legend(); ax[0].set_title("Time-series CV MAPE (%)")
ax[1].barh(["road", "weather", "congestion", "threat zone", "stops", "off-route/comm"][::-1], abt.set_index("제거한 변수군").reindex(list(groups))["증가"].values[::-1]); ax[1].set_title("MAPE increase when group removed")
imp.head(10)[::-1].plot.barh(ax=ax[2]); ax[2].set_title("Feature importance (top 10)")
plt.tight_layout(); plt.show()
In [ ]:
"""수송선단의 경로 이탈·불시 정차를 지속 시간과 이탈 거리로 신뢰도(0~1)를 매겨 경보 등급을 내는 감지 엔진을 만들어 이벤트 단위 재현율·오경보를 검증하고, 감지 신뢰도를 반영한 앙상블 ETA 모델이 기준 모델 대비 MAE·RMSE를 얼마나 줄이는지 Hold-out으로 평가한다."""
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor
from sklearn.metrics import r2_score
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 5] 수송선단 경로 이탈 및 불시 정차 감지 신뢰도 경보 엔진 개발 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 5] 수송선단 경로 이탈 및 불시 정차 감지 신뢰도 경보 엔진 개발 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(); d = add_features(df)
# ---------------- 1) 이탈·정차 신뢰도 점수 ----------------
def run_len(flag, trip):
"""같은 이동 안에서 조건이 연속으로 참인 횟수(현재 포함)."""
out, cur, last = [], 0, None
for f, t in zip(flag, trip):
cur = cur + 1 if (f and t == last) else (1 if f else 0)
out.append(cur); last = t
return np.array(out)
def score(off, speed):
"""이탈거리·속도 열에서 신뢰도(0~1)를 계산한다."""
dev_run = run_len((off >= 1.0).values, d["trip"].values)
stop_run = run_len((speed < 3).values, d["trip"].values)
return np.clip(off / 2.5, 0, 1) * (1 - 0.5 ** dev_run), 1 - np.exp(-stop_run * 5 / 10.0), dev_run, stop_run
d["dev_conf"], d["stop_conf"], d["dev_run"], d["stop_run"] = score(d["off_km"], d["speed"])
d["dev_true"] = (d["off_true"] >= 1.0).astype(int)
d["stop_ev"] = run_len((d["stopped_true"] == 1).values, d["trip"].values)
def event_table(dc, sc):
rows = []
for name, true_flag, conf in (("경로 이탈", d["dev_true"] == 1, dc), ("불시 정차(10분 이상)", d["stop_ev"] >= 2, sc)):
start = true_flag & ~true_flag.groupby(d["trip"]).shift(1, fill_value=False)
ev_ids = start.groupby(d["trip"]).cumsum().where(true_flag)
n_ev, hit, delays = 0, 0, []
for (tid, eid), g in d[true_flag].assign(ev=ev_ids[true_flag]).groupby(["trip", "ev"]):
n_ev += 1
a = g[conf[g.index] >= 0.5]
if len(a):
hit += 1; delays.append(a["t"].iloc[0] - g["t"].iloc[0])
alarm = conf >= 0.5
rows.append((name, n_ev, hit, hit / max(n_ev, 1), float(np.mean(delays)) if delays else np.nan, int(alarm.sum()), int((alarm & ~true_flag).sum())))
return pd.DataFrame(rows, columns=["유형", "실제 이벤트", "탐지", "재현율", "평균 탐지 지연(분)", "경보 관측", "오경보 관측"])
print("[1] 이벤트 단위 검증(신뢰도 0.5 이상을 경보로 간주) - 깨끗한 관측")
print(event_table(d["dev_conf"], d["stop_conf"]).round(2).to_string(index=False))
rng = np.random.default_rng(5)
noisy_off = d["off_km"] + np.abs(rng.normal(0, 0.9, len(d))) # 터널 음영·GPS 오차를 가정한 잡음
noisy_spd = (d["speed"] + rng.normal(0, 7, len(d))).clip(lower=0)
nd, ns, _, _ = score(noisy_off, noisy_spd)
print(" 잡음 조건(GPS 오차 +0.9km, 속도 오차 7km/h):")
print(event_table(pd.Series(nd, index=d.index), pd.Series(ns, index=d.index)).round(2).to_string(index=False))
for th in (0.3, 0.5, 0.8):
p = (d["stop_conf"] >= th); y = (d["stop_ev"] >= 2)
print(f" 불시 정차 임계값 {th}(깨끗한 관측): 정밀도 {float((p & y).sum() / max(p.sum(), 1)):.0%}, 관측 재현율 {float((p & y).sum() / max(y.sum(), 1)):.0%}")
print(" 잡음이 커지면 오경보가 늘고 이탈 판정이 흔들린다. 임계값은 경보 피로와 조기 대응 사이에서 업무적으로 정해야 한다.")
# ---------------- 3) 감지 신뢰도를 반영한 ETA 모델 vs 기준 모델 ----------------
FE2 = FEATS + ["dev_conf", "stop_conf", "dev_run", "stop_run"]
tr, te = d[d.trip < 160], d[(d.trip >= 160) & (d.eta_true >= 15)]
h1 = HistGradientBoostingRegressor(max_iter=200, max_depth=4, learning_rate=0.06, random_state=0).fit(tr[FE2], tr["eta_true"])
h0 = HistGradientBoostingRegressor(max_iter=150, max_depth=4, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["eta_true"])
rf = RandomForestRegressor(n_estimators=60, min_samples_leaf=5, max_features=0.5, random_state=0, n_jobs=1).fit(tr[FE2], tr["eta_true"])
P = {"기준(현재 속도 ETA)": te["base_eta"].values, "M4 모델": h0.predict(te[FEATS]), "HGB+감지 신뢰도": h1.predict(te[FE2]), "RandomForest": rf.predict(te[FE2])}
P["앙상블(평균)"] = (P["HGB+감지 신뢰도"] + P["RandomForest"]) / 2
tab = pd.DataFrame({k: {"MAE(분)": mae(te.eta_true, v), "RMSE(분)": float(np.sqrt(np.mean((te.eta_true - v) ** 2))), "MAPE(%)": mape(te.eta_true, v), "R2": r2_score(te.eta_true, v)} for k, v in P.items()}).T.round(3)
print(f"\n[2] Hold-out(이동 160~199, {len(te)}행) ETA 성능")
print(tab.to_string())
b, e, m4 = tab.loc["기준(현재 속도 ETA)"], tab.loc["앙상블(평균)"], tab.loc["M4 모델"]
imp_ok = 1 - e["MAE(분)"] / b["MAE(분)"] >= 0.2 and 1 - e["RMSE(분)"] / b["RMSE(분)"] >= 0.2
print(f" 기준 대비 MAE 개선 {1 - e['MAE(분)'] / b['MAE(분)']:.1%}, RMSE 개선 {1 - e['RMSE(분)'] / b['RMSE(분)']:.1%} (목표 20% 이상: {'달성' if imp_ok else '미달'}), R2 {e['R2']:.3f} (목표 0.85 이상: {'달성' if e['R2'] >= .85 else '미달'})")
print(f" 참고: M4 모델 대비 추가 개선은 MAE {1 - e['MAE(분)'] / m4['MAE(분)']:.1%}로, 감지 신뢰도 피처의 추가 효과는 작다(이미 정차 이력 피처가 있기 때문).")
stress = {"이탈 중": te.off_km >= 1, "정차 직후(경과 10분 이내)": (te.since_stop <= 10) & (te.stop_n > 0), "우천": te.rain == 1, "평상": (te.off_km < 1) & (te.stop_n == 0) & (te.rain == 0)}
rr = [(k, int(m.sum()), mae(te.eta_true[m], P["기준(현재 속도 ETA)"][m.values]), mae(te.eta_true[m], P["앙상블(평균)"][m.values])) for k, m in stress.items() if m.sum()]
print("[3] 불규칙 상황별 MAE(분): 기준 vs 앙상블")
print(pd.DataFrame(rr, columns=["조건", "건수", "기준", "앙상블"]).round(1).to_string(index=False))
print("경보는 판단 보조이며, 현장 확인과 대체 수송 지시는 사람(관제사·수송통제관)이 결정한다(M7).")
In [ ]:
"""실시간 수송선단 위치 추적과 ETA 변동 대시보드를 구현하고, ETA 지연분에 대한 요인군별 기여도 상위 3개를 건마다 자동 산출해 자연어 설명으로 보여 준다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import HistGradientBoostingRegressor
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 6] 실시간 수송선단 위치 추적 및 실시간 ETA 변동 대시보드 UI 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 6] 실시간 수송선단 위치 추적 및 실시간 ETA 변동 대시보드 UI 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(); d = add_features(df)
tr = d[d.trip < 160]
model = HistGradientBoostingRegressor(max_iter=150, max_depth=4, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["eta_true"])
d["eta_pred"] = model.predict(d[FEATS]).clip(min=0)
PLAN = 255 # 운행계획 소요시간(분, 가정)
# ---------------- 1) 요인군 기여도(평균 대체 방식): 거리 요인은 '순수 주행'으로 보고 지연 요인만 설명 ----------------
GROUPS = {"현재 속도": ["speed", "speed_ma3", "speed_std3"], "정차 이력": ["stop_n", "stop_min", "since_stop"], "우천": ["rain"], "시간대 정체": ["hour_sin", "hour_cos", "cong"],
"노선 구성(비포장·위협·교량터널)": ["rem_unpaved_km", "rem_threat_km", "rem_bridge_km"], "이탈·통신": ["off_km", "rssi"], "선단·화물": ["size", "k_냉장", "k_민감"]}
mean_row = tr[FEATS].mean()
te = d[d.trip >= 160].copy()
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["eta_pred"] - model.predict(x).clip(min=0) # 양수 = 이 요인이 도착을 늦춤
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) < 1.0).sum())
print(f"[1] ETA 예측 {len(te)}건 중 기여도 상위 3개 요인 자동 산출 {ok}건 ({ok / len(te):.0%}, 목표 100%), 3번째 요인 기여 1분 미만 {weak}건")
print(" 요인군별 평균 기여 크기(분):", ", ".join(f"{k} {v:.1f}" for k, v in contrib.abs().mean().sort_values(ascending=False).items()))
def explain(i):
r, c = te.loc[i], contrib.loc[i]
main = top3[i][0]
why = {"현재 속도": "현재 속도가 낮아", "정차 이력": "정차 이력으로", "우천": "우천으로", "시간대 정체": "시간대 정체로", "노선 구성(비포장·위협·교량터널)": "남은 구간(비포장·위협 구역)이 느려",
"이탈·통신": "노선 이탈·통신 상태로", "선단·화물": "선단 규모·화물 특성으로"}[main]
return f"trip {int(r['trip'])} t={int(r['t'])}분: {why} 도착까지 {r['eta_pred']:.0f}분 예상. 주요 요인 " + ", ".join(f"{g}({c[g]:+.1f}분)" for g in top3[i])
# ---------------- 2) 시점 스냅샷: 선단 12개의 실시간 상태판 ----------------
snap = te[(te["t"] == 120) & (te["trip"] < 172)].copy()
snap["도착 예정(분)"] = snap["t"] + snap["eta_pred"]
snap["계획 대비(분)"] = snap["도착 예정(분)"] - PLAN
snap["상태"] = np.select([snap["off_km"] >= 1, snap["is_stop"] == 1, snap["계획 대비(분)"] >= 20, snap["계획 대비(분)"] >= 10], ["이탈", "정차", "지연", "주의"], "정상")
print(f"\n[2] 선단 상태판(출발 후 120분, 선단 {len(snap)}개, 운행계획 {PLAN}분 기준) " + str(snap["상태"].value_counts().to_dict()))
show = snap[["trip", "progress", "speed", "eta_pred", "계획 대비(분)", "상태"]].rename(columns={"progress": "진행률", "speed": "속도", "eta_pred": "ETA(분)"}).round(1).head(8)
print(show.to_string(index=False))
worst = snap.sort_values("계획 대비(분)", ascending=False).head(3).index
print("[3] 지연 상위 3개 선단 자연어 설명")
for i in worst:
print(" -", explain(i))
# ---------------- 3) ETA 변동 추적(재생): 한 선단의 ETA가 시간에 따라 어떻게 바뀌나 ----------------
tid = int(te.loc[worst[0], "trip"])
q = te[te.trip == tid].reset_index(drop=True)
q["arrive_pred"] = q["t"] + q["eta_pred"]
q["change"] = q["arrive_pred"].diff()
alerts = q[q["change"].abs() >= 15]
arr_true = int(q["t"].iloc[-1] + q["eta_true"].iloc[-1])
print(f"\n[4] trip {tid} ETA 변동 추적: 실제 도착 {arr_true}분, 예측 도착 범위 {q['arrive_pred'].min():.0f}~{q['arrive_pred'].max():.0f}분, 15분 이상 변동 알림 {len(alerts)}회")
for _, r in alerts.head(3).iterrows():
print(f" t={int(r['t'])}분: 예측 도착 {r['arrive_pred']:.0f}분({r['change']:+.0f}분 변동), 속도 {r['speed']:.0f}km/h, 정차 누적 {int(r['stop_min'])}분")
print(" ETA 변동 알림은 판단 보조이며, 현장 확인과 대체 수송 여부는 사람이 결정한다(M7).")
fig, ax = plt.subplots(2, 2, figsize=(13, 8))
ax[0, 0].plot(PTS[0][0], PTS[0][1], "ks"); ax[0, 0].plot([p[0] for p in PTS], [p[1] for p in PTS], "-", c="lightgray", lw=6)
col = {"정상": "tab:green", "주의": "tab:olive", "지연": "tab:orange", "정차": "tab:red", "이탈": "tab:purple"}
for r in snap.itertuples():
ax[0, 0].plot(r.x, r.y, "o", c=col[r.상태], ms=9)
ax[0, 0].set_title("1 Convoy positions at t=120 min (green ok / red stopped / purple off-route)"); ax[0, 0].set_aspect("equal")
ax[0, 1].plot(q["t"], q["arrive_pred"], "o-", label="predicted arrival"); ax[0, 1].axhline(arr_true, c="k", ls="--", label="actual arrival"); ax[0, 1].axhline(PLAN, c="g", ls=":", label="plan")
ax[0, 1].legend(); ax[0, 1].set_title(f"2 ETA tracking, trip {tid} (minutes since departure)")
snap["상태"].value_counts().reindex(list(col)).fillna(0).plot.bar(ax=ax[1, 0], color=list(col.values())); ax[1, 0].set_title("3 Status count"); ax[1, 0].set_xlabel(""); ax[1, 0].set_xticklabels(["ok", "caution", "late", "stopped", "off-route"], rotation=0)
cw = contrib.loc[worst[0]]
ax[1, 1].barh(["speed", "stops", "rain", "congestion", "route mix", "off-route/comm", "fleet/cargo"][::-1], cw[list(GROUPS)].values[::-1]); ax[1, 1].set_title("4 Delay contribution of worst convoy (min)")
plt.suptitle("MOV06 convoy tracking dashboard (synthetic)"); plt.tight_layout(); plt.show()
In [ ]:
"""도착 지연 경보 발생 시 현장 확인·대체 수송 지시안을 비용 최적화로 추천하고(예비 대체 수송 수 제약 반영), 모든 지시를 관제사 상신 후 수송통제관·군수참모가 직접 승인·반려하는 HITL 절차(권한·2단계 결재·해시 체인 감사 로그)를 구현한다."""
import hashlib
import json
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.stats import norm
from sklearn.ensemble import HistGradientBoostingRegressor
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 7] 도착 지연 경보 발생 시 현장 확인 및 대체 수송 지시를 위한 HITL 절차 설계 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 7] 도착 지연 경보 발생 시 현장 확인 및 대체 수송 지시를 위한 HITL 절차 설계 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(); d = add_features(df)
tr = d[d.trip < 160]
model = HistGradientBoostingRegressor(max_iter=150, max_depth=4, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["eta_true"])
te = d[d.trip >= 160].copy()
te["eta_pred"] = model.predict(te[FEATS]).clip(min=0)
SIG = float(np.std(tr["eta_true"] - model.predict(tr[FEATS]))) # 예측 오차 표준편차(분)
# ---------------- 가정값(교육용): 운행계획, 지연 손실 단가, 조치별 비용·효과 ----------------
PLAN = 255 # 운행계획 소요시간(분)
W = {"일반": 0.5, "냉장": 1.5, "민감": 3.0} # 지연 1분당 손실(만원, 화물 구분별 가정)
C_CHECK, K_CHECK = 2.0, 0.75 # 현장 확인: 비용 2만원, 지연 25% 감소 가정
C_REPL, T_REPL = 45.0, 90 # 대체 수송: 비용 45만원, 지시 후 90분 뒤 도착
OBS = [60, 100, 140, 180] # 판단 시점(출발 후 분)
def exp_late(mu, sig):
"""지연(분) 기댓값 E[max(L,0)], L ~ N(mu, sig)."""
z = mu / sig
return sig * norm.pdf(z) + mu * norm.cdf(z)
def outcome(r, act):
"""실제 결과 비용(만원): 실제 도착 시각으로 평가."""
L = r["t"] + r["eta_true"] - PLAN
w = W[r["cargo"]]
if act == "대기":
return w * max(L, 0)
if act == "현장 확인":
return C_CHECK + K_CHECK * w * max(L, 0)
return C_REPL + w * max(r["t"] + T_REPL - PLAN, 0)
def decide(g, policy, reserve):
"""한 판단 시점의 선단 목록에 조치를 배정한다. policy: 'rule'(단순 ETA 10분 이상 지연이면 대체 출동) / 'opt'(기댓값 최소)."""
plan = {}
if policy == "rule":
for i, r in g.iterrows():
plan[i] = ("대체 수송", r["t"] + r["base_eta"] - PLAN) if r["t"] + r["base_eta"] - PLAN >= 10 else ("대기", 0)
else:
for i, r in g.iterrows():
mu, w = r["t"] + r["eta_pred"] - PLAN, W[r["cargo"]]
c = {"대기": w * exp_late(mu, SIG), "현장 확인": C_CHECK + K_CHECK * w * exp_late(mu, SIG),
"대체 수송": C_REPL + w * exp_late(r["t"] + T_REPL - PLAN, SIG)}
best = min(c, key=c.get)
plan[i] = (best, c["대기"] - c[best])
cand = sorted([i for i, (a, _) in plan.items() if a == "대체 수송"], key=lambda i: -plan[i][1])
for i in cand[reserve:]: # 예비 대체 수송 수 제약: 초과분은 '현장 확인'으로 낮춘다
plan[i] = ("현장 확인" if policy == "opt" else "대기", 0)
return {i: a for i, (a, _) in plan.items()}
def simulate(sub, policy, reserve):
cost, cnt = 0.0, {"대기": 0, "현장 확인": 0, "대체 수송": 0}
for t0 in OBS:
g = sub[(sub["t"] == t0) & (sub["eta_true"] > 0)]
for i, a in decide(g, policy, reserve).items():
cost += outcome(g.loc[i], a); cnt[a] += 1
return cost, cnt
# ---------------- 1) 최적화 시뮬레이터: 기존 관행 vs 기댓값 최소 추천 ----------------
SC = [("평상(예비 대체 수송 4대)", te, 4), ("예비 대체 수송 2대", te, 2), ("예비 대체 수송 0대", te, 0), ("우천 이동만", te[te.rain == 1], 4), ("민감 화물만", te[te.cargo == "민감"], 4)]
rows = []
for name, sub, R in SC:
cr, nr = simulate(sub, "rule", R)
co, no = simulate(sub, "opt", R)
rows.append((name, sum(nr.values()), cr, co, 1 - co / cr if cr > 0 else np.nan, nr["대체 수송"], no["대체 수송"], no["현장 확인"]))
res = pd.DataFrame(rows, columns=["시나리오", "판단 건수", "관행 비용(만원)", "추천 비용(만원)", "절감률", "관행 대체출동", "추천 대체출동", "추천 현장확인"])
print("[1] 지연 경보 조치 비용 비교(단가·효과는 교육용 가정값, 판단 시점 4회 x 선단)")
print(res.assign(절감률=res["절감률"].map("{:.1%}".format)).round(1).to_string(index=False))
base = res.iloc[0]
print(f" 평상 시나리오 절감 {base['절감률']:.1%} (목표 15% 이상: {'달성' if base['절감률'] >= .15 else '미달'}), 관행은 대체 출동 {int(base['관행 대체출동'])}회로 불필요한 출동이 많다")
print(" '관행'은 현재 속도 단순 ETA로 10분 이상 늦으면 무조건 대체 출동하는 가정 규칙이며 실제 부대 방침이 아니다.")
# ---------------- 2) 단가 민감도: 대체 수송 비용 ----------------
sens = []
for c in (25.0, 45.0, 70.0, 100.0):
C_REPL = c
cr, _ = simulate(te, "rule", 4); co, no = simulate(te, "opt", 4)
sens.append((c, 1 - co / cr, no["대체 수송"]))
C_REPL = 45.0
print("\n[2] 대체 수송 비용 민감도:", "; ".join(f"{c:.0f}만원 -> 절감 {s:.0%}, 추천 출동 {n}회" for c, s, n in sens))
print(" 비용이 높아질수록 추천은 현장 확인 중심으로 이동하고, 관행 대비 절감폭은 벌어진다.")
# ---------------- 3) HITL 결재 워크플로우 ----------------
USERS = {"ctl_a": "CTL", "ctl_b": "CTL", "tco_kim": "TCO", "tco_lee": "TCO", "log_park": "LOG", "aud_han": "AUD"}
PERM = {"CTL": {"submit", "redraft"}, "TCO": {"approve", "reject"}, "LOG": {"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, trip, action, detail, parent=None):
need = "redraft" if parent else "submit"
if need not in PERM.get(USERS.get(user), set()):
return self.deny(user, need, "-", "상신 권한 없음")
did = f"ALT{len(self.docs) + 1:02d}"
self.docs[did] = {"trip": trip, "action": action, "line": ["TCO", "LOG"] if action == "대체 수송" else ["TCO"], "step": 0, "status": "검토대기",
"drafter": user, "done": [], "parent": parent}
self._audit(user, "submit", did, f"trip {trip} {action} {detail}")
return did
def decide(self, user, did, act, reason=""):
role, doc = USERS.get(user), self.docs.get(did)
if role is None: return self.deny(user, act, did, "미등록 사용자")
if doc is None or doc["status"] != "검토대기": return self.deny(user, act, did, "결재 대상 아님(처리 완료)")
if act not in PERM[role]: return self.deny(user, act, did, f"{role} 역할은 {act} 불가")
if doc["line"][doc["step"]] != role: return self.deny(user, act, did, "결재 순서 위반")
if user == doc["drafter"] or user in doc["done"]: return self.deny(user, act, did, "기안자·기결재자 재결재 금지")
if len(reason.strip()) < 10: return self.deny(user, act, did, "사유 10자 이상 필수")
if act == "reject":
doc["status"] = "반려"
else:
doc["done"].append(user); doc["step"] += 1
doc["status"] = "승인" if doc["step"] == len(doc["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
g = te[(te["t"] == 140) & (te["eta_true"] > 0)]
act = decide(g, "opt", 3)
prop = [(i, a) for i, a in act.items() if a != "대기"]
prop = sorted(prop, key=lambda x: (x[1] != "대체 수송", x[0]))[:6]
wf = Flow()
ids = [wf.submit("ctl_a" if k % 2 == 0 else "ctl_b", int(g.loc[i, "trip"]), a, f"예상 지연 {g.loc[i, 't'] + g.loc[i, 'eta_pred'] - PLAN:+.0f}분") for k, (i, a) in enumerate(prop)]
two = [i for i in ids if len(wf.docs[i]["line"]) == 2]
print(f"\n[3] 140분 시점 AI 지시안 {len(ids)}건 전부 '검토대기' 등록, 자동 승인 {sum(wf.docs[i]['status'] == '승인' for i in ids)}건 (목표 0건). 2단계 결재(대체 수송) 대상 {two}")
first = two[0] if two else ids[0]
wf.decide("tco_kim", first, "reject", "예비 차량 위치 확인 후 재기안 요청, 근거 보강 필요")
new = wf.submit("ctl_a", wf.docs[first]["trip"], wf.docs[first]["action"], "재기안(예비 차량 정보 보강)", parent=first)
wf.decide("tco_kim", new, "approve", "예비 차량 가용 확인, 군수참모 상신함")
if wf.docs[new]["line"] == ["TCO", "LOG"]:
wf.decide("log_park", new, "approve", "대체 수송 시행, 도착 보고 조건")
for i in ids[1:3]:
wf.decide("tco_lee", i, "approve", "현장 확인 지시 타당, 결과 보고 조건")
tgt = ids[-1]
attacks = [("ctl_a", tgt, "approve", "내가 올린 건을 직접 승인한다"), ("aud_han", tgt, "approve", "감사 계정으로 승인 시도함"), ("ghost", tgt, "approve", "외부 계정으로 승인 시도함"),
("tco_kim", tgt, "approve", ""), ("tco_kim", tgt, "reject", "짧음"), ("log_park", tgt, "approve", "상급자가 순서를 건너뛰어 승인"),
("tco_lee", ids[1], "approve", "이미 승인된 건을 다시 승인함"), ("log_park", new, "reject", "승인된 건을 뒤집어 반려함"), ("ctl_b", tgt, "approve", "상신자 계정으로 승인 시도함")]
for u, doc, a, rs in attacks:
wf.decide(u, doc, a, rs)
wf.edit("ctl_a", new)
print(f" 권한 우회 시도 10건 중 차단 {len(wf.blocked)}건 (차단율 {len(wf.blocked) / 10:.0%}, 목표 100%)")
for u, a, doc, why in wf.blocked[:4]:
print(f" 차단 {u:<8} {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" {first}(반려) -> {new}(재기안, parent 연결) 이력:", [(r['action'], r['user']) for r in wf.log if r['doc'] in (first, 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="rule (always replace)"); ax[0].bar(x + .2, res["추천 비용(만원)"], .4, label="expected-cost optimum")
ax[0].set_xticks(x); ax[0].set_xticklabels([f"S{i + 1}" for i in x]); ax[0].legend(); ax[0].set_title("Total cost by scenario (10k KRW)")
ax[1].plot([s[0] for s in sens], [s[1] * 100 for s in sens], "o-"); ax[1].set_title("Saving (%) vs replacement cost"); ax[1].set_xlabel("replacement cost (10k KRW)")
plt.tight_layout(); plt.show()
In [ ]:
"""100개 수송선단의 위치 텔레메트리 스트림을 틱(5분) 단위로 모의 처리하는 엔진(중복 제거·손실·지연·GPS 잡음 대응, 이탈·불시 정차 경보, 배치 ETA 예측)의 대시보드 로딩 시간과 경보 정확도를 부하·오류 주입 시험으로 측정하고 처리 한계 규모를 도출한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import HistGradientBoostingRegressor
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 8] 100개 수송선단 동시 이동 모의 실시간 위치 데이터 스트림 부하 테스트 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 8] 100개 수송선단 동시 이동 모의 실시간 위치 데이터 스트림 부하 테스트 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(n_trips=100); d = add_features(df)
d["k"] = d.groupby("trip").cumcount()
K = int(d["k"].max()) + 1
def grid(col, fill=0.0):
return d.pivot(index="k", columns="trip", values=col).reindex(range(K)).fillna(fill).values
OFF, SPD, VALID = grid("off_km"), grid("speed"), grid("t", -1) >= 0
OFFT, STOPT = grid("off_true"), grid("stopped_true")
F = np.stack([grid(c) for c in FEATS], axis=-1).astype(np.float32) # [틱, 선단, 피처]
model = HistGradientBoostingRegressor(max_iter=100, max_depth=4, learning_rate=0.1, random_state=0).fit(d[FEATS][d.trip < 70].values, d["eta_true"][d.trip < 70].values)
def run_len2(flag):
"""열(선단)별로 조건이 연속 참인 횟수(현재 포함)를 틱 방향으로 센다."""
out, cur = np.zeros(flag.shape, int), np.zeros(flag.shape[1], int)
for k in range(flag.shape[0]):
cur = np.where(flag[k], cur + 1, 0); out[k] = cur
return out
TRUTH = (run_len2(OFFT >= 1.0) >= 2) | (run_len2(STOPT == 1) >= 2) # 정답: 이탈 또는 정차가 2회 연속 보고된 상태
print(f"[0] 스트림 원천: 선단 100개, 최대 {K}틱(5분 간격), 유효 보고 {int(VALID.sum())}건, 경보 상태 정답 {int((TRUTH & VALID).sum())}건({(TRUTH & VALID).sum() / VALID.sum():.1%})")
def run_stream(N, loss=0.0, dup=0.0, stale=0.0, noise=False, dedupe=True, seed=0, predict=True):
"""N개 선단의 스트림을 틱마다 일괄 처리한다. 반환: 틱 처리 지연(초) 리스트, 정확도 지표, 마지막 상태."""
rng = np.random.default_rng(seed)
idx = np.arange(N) % 100
dev_run, stop_run = np.zeros(N, int), np.zeros(N, int)
prev_o, prev_s, eta = np.zeros(N), np.full(N, 50.0), np.zeros(N)
lat, tp, fp, fn, tn, n_dup_ignored = [], 0, 0, 0, 0, 0
for k in range(K):
t0 = time.perf_counter()
valid = VALID[k, idx]
recv = valid & (rng.random(N) >= loss) # 패킷 손실이면 갱신 없음(상태 유지)
dups = recv & (rng.random(N) < dup) # 같은 보고가 두 번 도착
o, s = OFF[k, idx].copy(), SPD[k, idx].copy()
if stale > 0:
old = rng.random(N) < stale
o, s = np.where(old, prev_o, o), np.where(old, prev_s, s) # 지연 도착: 이전 틱 값이 도착
if noise:
o = o + np.abs(rng.normal(0, 0.9, N)); s = np.clip(s + rng.normal(0, 7, N), 0, None)
prev_o, prev_s = OFF[k, idx], SPD[k, idx]
times = recv.astype(int) if dedupe else recv.astype(int) + dups.astype(int)
n_dup_ignored += int(dups.sum()) if dedupe else 0
dev_run = np.where(recv, np.where(o >= 1.0, dev_run + times, 0), dev_run)
stop_run = np.where(recv, np.where(s < 3, stop_run + times, 0), stop_run)
alarm = (dev_run >= 2) | (stop_run >= 2)
if predict and recv.any():
eta[recv] = model.predict(F[k, idx[recv]])
lat.append(time.perf_counter() - t0)
tr_, v = TRUTH[k, idx], valid
tp += int((alarm & tr_ & v).sum()); fp += int((alarm & ~tr_ & v).sum()); fn += int((~alarm & tr_ & v).sum()); tn += int((~alarm & ~tr_ & v).sum())
tot = tp + fp + fn + tn
return lat, {"정확도": (tp + tn) / tot, "재현율": tp / max(tp + fn, 1), "오경보": fp, "미탐": fn, "중복무시": n_dup_ignored}, (alarm, eta, idx)
def dashboard(alarm, eta, idx):
"""대시보드 첫 화면: 상태 집계 + 지연 상위 10개 + 차트 렌더링(Agg)까지 걸린 시간을 잰다."""
t0 = time.perf_counter()
tb = pd.DataFrame({"convoy": np.arange(len(idx)), "eta": eta, "alarm": alarm})
summ = tb["alarm"].value_counts().to_dict()
top = tb.sort_values("eta", ascending=False).head(10)
fg, ax = plt.subplots(1, 2, figsize=(9, 4)) # 위치 산점도와 ETA 막대를 실제로 그려 렌더링 시간을 포함
ax[0].scatter(np.arange(len(tb)) % 100, eta, c=np.where(alarm, "r", "g"), s=4); ax[1].barh(top["convoy"].astype(str), top["eta"])
fg.canvas.draw(); plt.close(fg)
return time.perf_counter() - t0, summ, top
# ---------------- 1) 규모별 부하 시험 ----------------
rows = []
for N in (100, 300, 1000, 3000, 10000):
lat, acc, st = run_stream(N, loss=0.05, dup=0.05)
dl, _, _ = dashboard(*st)
lat = np.array(lat) * 1000
rows.append((N, np.median(lat), np.percentile(lat, 95), lat.max(), N / (np.median(lat) / 1000), dl, acc["정확도"]))
ld = pd.DataFrame(rows, columns=["선단 수", "틱 처리 중앙값(ms)", "p95(ms)", "최대(ms)", "처리량(건/초)", "대시보드 로딩(초)", "경보 정확도"])
print("\n[1] 규모별 부하 시험(손실 5%, 중복 5% 포함, 틱마다 ETA 배치 예측 포함)")
print(ld.round(3).to_string(index=False))
r100 = ld.iloc[0]
print(f" 100개 선단: 대시보드 로딩 {r100['대시보드 로딩(초)']:.3f}초(목표 2초 이내: {'달성' if r100['대시보드 로딩(초)'] <= 2 else '미달'}), 틱 p95 {r100['p95(ms)']:.0f}ms (5분 보고 주기 대비 충분)")
lim = ld[(ld["p95(ms)"] <= 2000) & (ld["대시보드 로딩(초)"] <= 2)]["선단 수"].max()
print(f" 틱 p95 2초·대시보드 2초 이내를 모두 지킨 최대 시험 규모: {int(lim)}개 선단 (이 환경 기준, 실제 서버는 별도 시험 필요)")
# ---------------- 2) 오류 주입: 경보 정확도 ----------------
CASES = [("정상", {}), ("패킷 손실 15%", dict(loss=.15)), ("중복 20%(중복 제거 켬)", dict(dup=.2)), ("중복 20%(중복 제거 끔)", dict(dup=.2, dedupe=False)),
("지연 도착 10%", dict(stale=.1)), ("GPS·속도 잡음", dict(noise=True)), ("손실+중복+지연 복합", dict(loss=.1, dup=.1, stale=.1))]
res = []
for name, kw in CASES:
accs = [run_stream(100, seed=s, predict=False, **kw)[1] for s in range(5)]
res.append((name, np.mean([a["정확도"] for a in accs]), np.mean([a["재현율"] for a in accs]), np.mean([a["오경보"] for a in accs]), np.mean([a["미탐"] for a in accs])))
er = pd.DataFrame(res, columns=["조건", "정확도", "재현율", "오경보(평균)", "미탐(평균)"])
print("\n[2] 오류 주입 시 경보 정확도(100개 선단, 시드 5회 평균, 정확도=틱 단위 경보 상태 일치율)")
print(er.assign(정확도=er["정확도"].map("{:.1%}".format), 재현율=er["재현율"].map("{:.1%}".format)).round(1).to_string(index=False))
ok_set = er[er["조건"].isin(["정상", "패킷 손실 15%", "중복 20%(중복 제거 켬)", "지연 도착 10%"])]["정확도"].min()
print(f" 통신 오류 4개 조건 최저 정확도 {ok_set:.1%} (목표 95% 이상: {'달성' if ok_set >= .95 else '미달'}). 중복 제거를 끄면 오경보가 {er.iloc[3]['오경보(평균)']:.0f}건으로 늘고, 잡음 조건은 정확도 {er.iloc[5]['정확도']:.1%}·재현율 {er.iloc[5]['재현율']:.0%}로 떨어진다.")
print(" 패킷 손실 조건은 정확도는 높아도 재현율이 낮다(경보 상태가 희소해 정확도가 부풀려지므로 재현율·오경보를 함께 본다).")
print(" 잡음 조건은 GPS 오차를 가정한 최악 시험이며, 오경보가 늘어 임계값·연속 횟수 조정 또는 보조 센서 확인이 필요하다.")
# ---------------- 3) 한계치와 대응 ----------------
lat, acc, st = run_stream(10000, loss=0.2, dup=0.2, stale=0.1)
dl, summ, top = dashboard(*st)
print(f"\n[3] 한계 시험(선단 10000개, 손실 20%+중복 20%+지연 10%): 틱 p95 {np.percentile(lat, 95) * 1000:.0f}ms, 대시보드 {dl:.3f}초, 경보 상태 {summ}")
print(" 대시보드는 전체 표 대신 요약·지연 상위 10개·경보 목록만 먼저 보여 주어 로딩을 줄인다. 규모가 커지면 화면 갱신을 틱 단위 증분 갱신으로 바꾼다.")
print(" 경보는 판단 보조이며 현장 확인과 대체 수송은 사람이 결정한다(M7). 모든 선단·좌표는 교육용 가상 자료이다.")
fig, ax = plt.subplots(1, 3, figsize=(16, 4.2))
ax[0].loglog(ld["선단 수"], ld["p95(ms)"], "o-", label="tick p95"); ax[0].axhline(2000, c="r", ls="--", label="2 s limit"); ax[0].legend(); ax[0].set_title("Tick latency vs convoys (ms)")
ax[1].barh([f"C{i + 1}" for i in range(len(er))][::-1], er["정확도"].values[::-1] * 100); ax[1].axvline(95, c="r", ls="--"); ax[1].set_xlim(80, 100.5); ax[1].set_title("Alarm accuracy by fault case (%)")
ax[2].plot(ld["선단 수"], ld["대시보드 로딩(초)"], "s-"); ax[2].set_xscale("log"); ax[2].set_title("Dashboard load time (s)")
plt.tight_layout(); plt.show()
In [ ]:
"""AAR로 ETA 예측 오차를 조건별·사례별로 분석해 원인을 분류하고, 지연 요인 가중치 매트릭스(분/km 단위)를 비음수 최소제곱으로 재튜닝해 보류 기간에서 개선을 검증한 뒤, 비용 절감액·시간 단축률·투자 회수 기간 3대 효과 지표를 산식과 민감도로 산출한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.optimize import nnls
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 9] AAR 기반 ETA 예측 오차 분석 및 지연 요인 가중치 매트릭스 재튜닝 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 9] AAR 기반 ETA 예측 오차 분석 및 지연 요인 가중치 매트릭스 재튜닝 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
df = make_telemetry(); d = add_features(df)
d = d[d.eta_true >= 15].copy()
d["unp_share"] = d["rem_unpaved_km"] / d["remaining_km"].clip(lower=1)
d["thr_share"] = d["rem_threat_km"] / d["remaining_km"].clip(lower=1)
d["brg_share"] = d["rem_bridge_km"] / d["remaining_km"].clip(lower=1)
d["peak"] = 1 - d["cong"]
# 가중치 매트릭스: 남은 1km당 소요(분) = 기본 + 요인별 가산. 요인 값은 관측 시점의 상황.
COLS = {"기본(분/km)": None, "비포장 비율": "unp_share", "위협 구역 비율": "thr_share", "교량·터널 비율": "brg_share", "우천": "rain", "우천 x 비포장": None, "시간대 정체": "peak", "선단 규모(대)": "size"}
NAMES = list(COLS)
def design(x):
km = x["remaining_km"].values[:, None]
z = np.column_stack([np.ones(len(x)), x["unp_share"], x["thr_share"], x["brg_share"], x["rain"], x["rain"] * x["unp_share"], x["peak"], x["size"] / 10])
return km * z
def eta_rule(x, w):
return design(x) @ w
W0 = np.array([1.00, 0.50, 0.30, 0.10, 0.10, 0.10, 0.40, 0.05]) # 도입 시점(과거 기록 부족) 전문가 추정 가중치
past, ops, hold = d[d.trip < 100], d[(d.trip >= 100) & (d.trip < 150)], d[d.trip >= 150]
# ---------------- 1) AAR 오차 분석: 운영 구간(이동 100~149) ----------------
ops = ops.assign(pred=eta_rule(ops, W0), err=lambda x: x["pred"] - x["eta_true"])
print(f"[1] AAR 대상: 이동 100~149({ops.trip.nunique()}개, {len(ops)}행) 현행 가중치 ETA MAPE {mape(ops.eta_true, ops.pred):.1f}%, MAE {mae(ops.eta_true, ops.pred):.1f}분, 평균 편향 {ops.err.mean():+.1f}분(음수=과소 예측)")
conds = {"우천": ops.rain == 1, "시간대 정체": ops.peak > 0, "정차 중/정차 직후": (ops.since_stop <= 10) & (ops.stop_n > 0), "비포장 구간 위주": ops.unp_share > 0.5, "냉장 화물": ops.cargo == "냉장", "민감 화물": ops.cargo == "민감", "평상": (ops.rain == 0) & (ops.peak == 0) & (ops.stop_n == 0)}
ct = pd.DataFrame([(k, int(m.sum()), mape(ops.eta_true[m], ops.pred[m]), ops.err[m].mean()) for k, m in conds.items() if m.sum()], columns=["조건", "건수", "MAPE(%)", "평균 편향(분)"])
print(ct.round(1).to_string(index=False))
worst = ops.reindex(ops.err.abs().sort_values(ascending=False).index).head(40)
def cause(r):
if r["stop_n"] > 0 and r["since_stop"] <= 10: return "불시 정차 직후"
if r["rain"] == 1: return "우천 과소 반영"
if r["peak"] > 0: return "시간대 정체 과소 반영"
if r["unp_share"] > 0.5: return "비포장 구간 과소 반영"
return "기타(속도 변동)"
cs = worst.apply(cause, axis=1).value_counts()
print(" 오차 큰 상위 40건의 원인 분류:", ", ".join(f"{k} {v}건" for k, v in cs.items()))
# ---------------- 2) 가중치 재튜닝: 비음수 최소제곱(운영 구간 학습 -> 보류 구간 검증) ----------------
w1, _ = nnls(design(ops), ops["eta_true"].values)
ho = hold.assign(p0=eta_rule(hold, W0), p1=eta_rule(hold, w1))
b = (mape(ho.eta_true, ho.p0), mae(ho.eta_true, ho.p0), np.sqrt(np.mean((ho.eta_true - ho.p0) ** 2)))
a = (mape(ho.eta_true, ho.p1), mae(ho.eta_true, ho.p1), np.sqrt(np.mean((ho.eta_true - ho.p1) ** 2)))
mt = pd.DataFrame({"재튜닝 전": W0, "재튜닝 후": w1}, index=NAMES)
mt["변화율"] = np.where(mt["재튜닝 전"] > 0, mt["재튜닝 후"] / mt["재튜닝 전"] - 1, np.nan)
print("\n[2] 지연 요인 가중치 매트릭스(남은 1km당 분)")
print(mt.round(3).to_string())
print(f" 보류 구간(이동 150~199, {len(ho)}행): MAPE {b[0]:.1f}% -> {a[0]:.1f}%, MAE {b[1]:.1f} -> {a[1]:.1f}분, RMSE {b[2]:.1f} -> {a[2]:.1f}분")
chg = mt[(mt["변화율"].abs() > 0.5)].index.tolist()
print(f" 변화율 50% 초과 요인 {chg or '없음'}는 사람이 검토·승인한 뒤 적용한다(자동 반영 금지). 학습 기간(이동 0~99) 재현: 재튜닝 후 MAPE {mape(past.eta_true, eta_rule(past, w1)):.1f}%")
print(" 우천과 비포장 가중치가 크게 바뀌면 현장 관측 근거를 함께 확인해야 하며, 노선이 하나뿐인 가상 자료라 실제 노선에는 다시 학습해야 한다.")
# ---------------- 3) ROI 3대 지표(교육용 가정값) ----------------
TRIPS_YR, STAFF, WAGE, MARGIN = 1200, 6, 1.2, 15 # 연 수송 횟수, 수령 대기 인원, 시간당 인건비(만원), 안전 여유 시간(분)
INV, OPEX = 3000.0, 120.0 # 도입 비용(만원), 연 운영비(만원)
def roi(mae_b, mae_a, trips=TRIPS_YR, inv=INV):
wait_b, wait_a = MARGIN + mae_b, MARGIN + mae_a # 수령 측은 ETA 오차만큼 추가로 대기·준비한다고 가정
save = (wait_b - wait_a) / 60 * STAFF * WAGE * trips - OPEX
return save, 1 - wait_a / wait_b, inv / max(save / 12, 1e-9)
sv, rate, pb = roi(b[1], a[1])
print(f"\n[3] ROI 3대 지표 [교육용 가정: 연 {TRIPS_YR}회, 대기 인원 {STAFF}명, 인건비 {WAGE}만원/시, 도입 {INV:.0f}만원, 운영 {OPEX:.0f}만원/년]")
print(f" (1) 연간 비용 절감액 {sv:,.0f}만원 (2) 수령 대기·준비 시간 단축률 {rate:.1%} (3) 투자 회수 기간 {pb:.1f}개월")
print(" 산식: 절감액 = (재튜닝 전 대기 - 후 대기)/60 x 인원 x 인건비 x 횟수 - 운영비, 대기 = 안전 여유 15분 + ETA 평균 오차")
sens = [(t, roi(b[1], a[1], trips=t)) for t in (300, 600, 1200, 2400)]
print(" 횟수 민감도:", "; ".join(f"{t}회 -> 절감 {s[0]:,.0f}만원, 회수 {s[2]:.0f}개월" for t, s in sens))
print(" 비관 시나리오(오차 개선 절반만 실현, 400회):", f"절감 {roi(b[1], (a[1] + b[1]) / 2, trips=400)[0]:,.0f}만원, 회수 {roi(b[1], (a[1] + b[1]) / 2, trips=400)[2]:.0f}개월")
print(" 효과 3지표 산출 완료(비용 절감액·시간 단축률·회수 기간). 가정값이므로 재정·기획 부서와 실제 단가로 다시 산정해야 한다.")
fig, ax = plt.subplots(1, 3, figsize=(16, 4.3))
ax[0].bar(ct["조건"].index, ct["MAPE(%)"]); ax[0].set_xticks(range(len(ct))); ax[0].set_xticklabels([f"C{i + 1}" for i in range(len(ct))]); ax[0].set_title("AAR: MAPE (%) by condition")
x = np.arange(len(mt)); ax[1].bar(x - .2, mt["재튜닝 전"], .4, label="before"); ax[1].bar(x + .2, mt["재튜닝 후"], .4, label="after"); ax[1].set_xticks(x); ax[1].set_xticklabels([f"w{i}" for i in x]); ax[1].legend(); ax[1].set_title("Delay weight matrix (min/km)")
ax[2].scatter(ho.eta_true, ho.p0, s=6, alpha=.4, label="before"); ax[2].scatter(ho.eta_true, ho.p1, s=6, alpha=.4, label="after"); ax[2].plot([0, 250], [0, 250], "k--"); ax[2].legend(); ax[2].set_title("Hold-out ETA: actual vs predicted (min)")
plt.tight_layout(); plt.show()
In [ ]:
"""텔레메트리 갱신, 정합성 검증, 드리프트 점검, 자동 재학습(챔피언/챌린저), 선단 ETA 예측과 지연·이탈·정차 경보, 승인대기 분리, 7일 납품 소요 예측까지 한 번에 실행하는 통합 파이프라인을 완성하고 운영 문서 완비율·자체 점검·인터페이스 명세·브리핑 요약을 출력한다."""
import importlib
import json
import sys
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.linear_model import Ridge
# ---------------------------------------------------------------------------
# [교육용가상자료] 가상 수송 노선·수송선단 텔레메트리 생성기 (실제 도로·지명·부대 없음)
# 가상 노선(출발 거점 -> 도착 거점, 5개 구간)을 5분 간격으로 보고하는 선단 GPS·화물 센서·통신 상태 자료를 만든다.
# ---------------------------------------------------------------------------
PTS = [(0, 0), (30, 12), (62, 14), (88, 38), (112, 42), (135, 66)] # 노선 꼭짓점(km, 가상 좌표)
ROUTE = pd.DataFrame({"seg": ["R1", "R2", "R3", "R4", "R5"], "road": ["paved", "unpaved", "paved", "paved", "unpaved"],
"limit": [70, 50, 70, 50, 50], "kind": ["일반", "일반", "일반", "교량·터널", "일반"], "threat": [0, 0, 1, 0, 0]})
ROUTE["x0"], ROUTE["y0"] = [p[0] for p in PTS[:-1]], [p[1] for p in PTS[:-1]]
ROUTE["x1"], ROUTE["y1"] = [p[0] for p in PTS[1:]], [p[1] for p in PTS[1:]]
ROUTE["km"] = np.hypot(ROUTE["x1"] - ROUTE["x0"], ROUTE["y1"] - ROUTE["y0"])
ROUTE["c0"] = np.r_[0, np.cumsum(ROUTE["km"])[:-1]]
ROUTE["c1"] = np.cumsum(ROUTE["km"])
LEN = float(ROUTE["km"].sum())
ROAD_F = {"paved": 1.0, "unpaved": 0.75}
C1L, C0L, KML = ROUTE["c1"].tolist(), ROUTE["c0"].tolist(), ROUTE["km"].tolist() # 반복 계산용 리스트(속도)
BASE_V = [l * ROAD_F[r] for l, r in zip(ROUTE["limit"], ROUTE["road"])]
THR, UNP, TUN = ROUTE["threat"].tolist(), (ROUTE["road"] == "unpaved").tolist(), (ROUTE["kind"] == "교량·터널").tolist()
def seg_of(s):
# [TODO : 학생 실습 10] 군수품 이동 상태 추적 및 도착 예정 시간(ETA) 예측 시스템 최종 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def route_xy(s):
"""노선 위 진행거리 s(km)의 좌표와 진행 방향(도)."""
# [TODO : 학생 실습 10] 군수품 이동 상태 추적 및 도착 예정 시간(ETA) 예측 시스템 최종 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def project(x, y):
"""GPS 좌표를 노선에 투영해 (진행거리 km, 노선과의 수직 이탈 km)를 구한다."""
x, y = np.atleast_1d(x).astype(float), np.atleast_1d(y).astype(float)
best_s, best_d = np.zeros(len(x)), np.full(len(x), np.inf)
for r in ROUTE.itertuples():
dx, dy = r.x1 - r.x0, r.y1 - r.y0
f = np.clip(((x - r.x0) * dx + (y - r.y0) * dy) / (dx * dx + dy * dy), 0, 1)
d = np.hypot(x - (r.x0 + f * dx), y - (r.y0 + f * dy))
better = d < best_d
best_s, best_d = np.where(better, r.c0 + f * r.km, best_s), np.where(better, d, best_d)
return best_s, best_d
def cong(clock_min):
h = (clock_min / 60) % 24
return 0.82 if (7 <= h < 9 or 17 <= h < 19) else 1.0 # 출퇴근 시간대 정체 계수(가정)
def simulate_trip(rng, tid, start_min, rain, size, cargo):
"""1분 단위로 이동을 모의하고 5분마다 텔레메트리를 남긴다. 이탈·불시 정차 이벤트는 정답 라벨로 함께 저장한다."""
s, t, stop_left, rows = 0.0, 0, 0, []
dev = (rng.random() < 0.12)
dev_t0 = int(rng.integers(30, 120)) if dev else -1
dev_len = int(rng.integers(12, 25)) if dev else 0
cause = ""
while s < LEN:
i = seg_of(s)
v = BASE_V[i] * (0.88 if rain else 1.0) * (0.85 if (rain and UNP[i]) else 1.0) * cong(start_min + t) \
* (0.7 if THR[i] else 1.0) * (1 - 0.012 * size) * rng.normal(1, 0.06)
off = 0.0
if dev and dev_t0 <= t < dev_t0 + dev_len:
ph = (t - dev_t0) / dev_len
off, v = 4.0 * (1 - abs(2 * ph - 1)), v * 0.4 # 노선 이탈(삼각형 형태로 최대 4km)
if stop_left > 0:
v, stop_left = 0.0, stop_left - 1
elif rng.random() < 0.0015:
stop_left, cause = int(rng.integers(8, 36)), ("maint" if rng.random() < 0.7 else "urgent")
s += v / 60
t += 1
if t % 5 == 0 or s >= LEN:
x, y, hd = route_xy(min(s, LEN))
nx, ny = np.cos(np.radians(hd)), -np.sin(np.radians(hd))
j = seg_of(min(s, LEN)); tun = TUN[j]
rows.append(dict(trip=tid, t=t, clock=(start_min + t) % 1440, x=x + off * nx + rng.normal(0, 0.03), y=y + off * ny + rng.normal(0, 0.03),
speed=max(v * rng.normal(1, 0.03), 0.0), heading=(hd + rng.normal(0, 3)) % 360, rssi=-62 - 14 * tun + rng.normal(0, 3),
cargo_temp=18 + 3 * np.sin(t / 40) + rng.normal(0, 0.3), shock_g=abs(rng.normal(0.25, 0.1)) * (2.0 if UNP[j] else 1.0),
s_true=min(s, LEN), off_true=off, stopped_true=int(v == 0.0), rain=int(rain), size=size, cargo=cargo, stop_cause=cause if v == 0.0 else ""))
arr = t
for r in rows:
r["eta_true"] = arr - r["t"]
return rows
def make_telemetry(seed=21, n_trips=200):
"""n_trips개 수송선단 이동 기록(날짜 순). 선단 규모 3~12대, 화물 구분 일반/냉장/민감."""
rng = np.random.default_rng(seed)
rows = []
for k in range(n_trips):
rows += simulate_trip(rng, k, int(rng.integers(5 * 60, 15 * 60)), bool(rng.random() < 0.2), int(rng.integers(3, 13)), str(rng.choice(["일반", "냉장", "민감"], p=[.6, .25, .15])))
return pd.DataFrame(rows)
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))))
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# ETA 예측 피처: 관측 시점까지의 텔레메트리와 남은 노선 구성(포장·위협·교량터널 구간 거리)만 사용
# ---------------------------------------------------------------------------
FEATS = ["remaining_km", "progress", "speed", "speed_ma3", "speed_std3", "stop_n", "stop_min", "since_stop", "hour_sin", "hour_cos", "cong", "rain", "size",
"k_냉장", "k_민감", "rem_unpaved_km", "rem_threat_km", "rem_bridge_km", "off_km", "rssi", "base_eta"]
def rem_composition(s):
"""진행거리 s 이후 남은 구간 중 비포장·위협·교량터널 거리(km)."""
left = np.clip(ROUTE["c1"].values[None, :] - np.maximum(np.atleast_1d(s)[:, None], ROUTE["c0"].values[None, :]), 0, None)
return left @ np.array(UNP, float), left @ np.array(THR, float), left @ np.array(TUN, float)
def add_features(df):
d = df.sort_values(["trip", "t"]).copy()
d["s_obs"], d["off_km"] = project(d["x"].values, d["y"].values)
d["remaining_km"] = LEN - d["s_obs"]
d["progress"] = d["s_obs"] / LEN
g = d.groupby("trip")["speed"]
d["speed_ma3"] = g.rolling(3, min_periods=1).mean().reset_index(level=0, drop=True)
d["speed_std3"] = g.rolling(3, min_periods=1).std().reset_index(level=0, drop=True).fillna(0)
d["is_stop"] = (d["speed"] < 3).astype(int)
new_stop = (d["is_stop"] == 1) & (d.groupby("trip")["is_stop"].shift(1).fillna(0) == 0)
d["stop_n"] = new_stop.astype(int).groupby(d["trip"]).cumsum()
d["stop_min"] = d.groupby("trip")["is_stop"].cumsum() * 5
blk = (d["is_stop"] == 1).astype(int).groupby(d["trip"]).cumsum() # 정차 누적 횟수로 구간을 나누어 마지막 정차 이후 경과 계산
d["since_stop"] = d.groupby(["trip", blk]).cumcount() * 5
d["hour_sin"], d["hour_cos"] = np.sin(2 * np.pi * d["clock"] / 1440), np.cos(2 * np.pi * d["clock"] / 1440)
d["cong"] = d["clock"].map(cong)
d["k_냉장"], d["k_민감"] = (d["cargo"] == "냉장").astype(int), (d["cargo"] == "민감").astype(int)
d["rem_unpaved_km"], d["rem_threat_km"], d["rem_bridge_km"] = rem_composition(d["s_obs"].values)
d["base_eta"] = np.minimum(d["remaining_km"] / np.maximum(d["speed"], 5.0) * 60, 600) # 현재 속도 기반 단순 ETA(분, 정차 시 폭주 방지로 5km/h 하한·600분 상한)
return d.reset_index(drop=True)
# ---------------------------------------------------------------------------
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_telemetry(n_trips=240) # 이동 0~199 기존 + 200~239 신규 입력분
d = add_features(raw)
chk = {"(이동,시각) 유일": not raw.duplicated(["trip", "t"]).any(), "결측 0": int(raw.isna().sum().sum()) == 0,
"좌표 범위(가상 영역)": bool(raw.x.between(-10, 150).all() and raw.y.between(-10, 80).all()), "속도 0~120km/h": bool(raw.speed.between(0, 120).all()),
"방향 0~360": bool(raw.heading.between(0, 360).all()), "ETA 비음수": bool((raw.eta_true >= 0).all())}
print(f"[1] 신규 40개 이동 포함 {len(raw)}행({raw.trip.nunique()}개 이동) 정합성 검증 {sum(chk.values())}/{len(chk)} 통과")
# ---------------- 2) 자동 재학습: 드리프트 점검 -> 챌린저 학습 -> 승격 판단 ----------------
def fit(tr):
return HistGradientBoostingRegressor(max_iter=150, max_depth=4, learning_rate=0.08, random_state=0).fit(tr[FEATS], tr["eta_true"])
def evaluate(m, te):
te = te[te.eta_true >= 15]
p = m.predict(te[FEATS]).clip(min=0)
return mae(te["eta_true"], p), mape(te["eta_true"], p)
def retrain_pipeline(old_end, new_end, hold_n=10, drift_mape=8.0, min_gain=0.03):
champ = fit(d[d.trip < old_end])
new = d[(d.trip >= old_end) & (d.trip < new_end)]
ch_mae, ch_mape = evaluate(champ, new)
hold = d[(d.trip >= new_end - hold_n) & (d.trip < new_end)]
chall = fit(d[d.trip < new_end - hold_n])
a, b = evaluate(champ, hold)[0], evaluate(chall, hold)[0]
promote = (a - b) / a >= min_gain
return {"champ_mape": ch_mape, "drift": ch_mape > drift_mape, "champ_mae": a, "chall_mae": b, "promote": promote, "model": chall if promote else champ}
rt = retrain_pipeline(160, 240)
print(f"[2] 재학습 파이프라인: 기존 모델의 신규 80개 이동 MAPE {rt['champ_mape']:.2f}% (드리프트 기준 8.0%: {'초과' if rt['drift'] else '이내'})")
print(f" 검증 10개 이동 MAE 챔피언 {rt['champ_mae']:.2f}분 vs 챌린저 {rt['chall_mae']:.2f}분 -> {'챌린저 승격' if rt['promote'] else '기존 모델 유지'}(개선 3% 이상일 때만 승격)")
model = rt["model"]
# ---------------- 3) 선단 ETA 예측과 경보(승인대기 분리) ----------------
PLAN = 255
live = d[(d.trip >= 230) & (d.t == 120) & (d.eta_true > 0)].copy()
live["eta_pred"] = model.predict(live[FEATS]).clip(min=0)
live["계획 대비(분)"] = live["t"] + live["eta_pred"] - PLAN
live["is_stop"] = (live["speed"] < 3).astype(int)
live["경보"] = np.select([live["off_km"] >= 1, live["is_stop"] == 1, live["계획 대비(분)"] >= 20], ["이탈 의심", "정차", "지연 20분 이상"], "정상")
live["처리"] = np.where(live["경보"] != "정상", "관제사 확인 후 수송통제관 승인대기", "추적 유지")
print(f"\n[3] 출발 후 120분 선단 {len(live)}개 상태판(운행계획 {PLAN}분) [교육용가상자료]")
print(live[["trip", "progress", "speed", "eta_pred", "계획 대비(분)", "경보", "처리"]].rename(columns={"progress": "진행률", "speed": "속도", "eta_pred": "ETA(분)"}).round(1).to_string(index=False))
pend = live[live["처리"].str.contains("승인대기")]
print(f" 경보 {len(pend)}건은 자동으로 대체 수송을 지시하지 않고 승인대기로 분리한다(미션 7).")
# ---------------- 4) 7일 납품 소요 예측(요일 Ridge vs 전주 동요일 Baseline) ----------------
rg_ = np.random.default_rng(31)
days = np.arange(180)
dem = (40 + 6 * np.sin(2 * np.pi * days / 7) + 0.05 * days + np.where((days >= 120) & (days < 130), 12, 0) + rg_.normal(0, 2, len(days))).round() # 일 납품 수송 소요(가상, 대)
def design(dd):
dd = np.asarray(dd)
return np.column_stack([np.sin(2 * np.pi * dd / 7), np.cos(2 * np.pi * dd / 7), dd / 100, np.ones(len(dd)), ((dd >= 120) & (dd < 130)).astype(float)])
tr_d, te_d = days[:158], days[158:]
rg = Ridge(alpha=0.1, fit_intercept=False).fit(design(tr_d), dem[tr_d])
f_r, f_n = rg.predict(design(te_d)), dem[te_d - 7]
print(f"\n[4] 납품 소요 예측 검증(day 158~179, 22일): Ridge MAPE {mape(dem[te_d], f_r):.1f}% vs 전주 동요일 {mape(dem[te_d], f_n):.1f}%")
nxt = np.arange(180, 187)
rg = Ridge(alpha=0.1, fit_intercept=False).fit(design(days), dem)
fc = rg.predict(design(nxt))
print(" 향후 7일 납품 소요 예측(대):", ", ".join(f"{int(v)}" for v in fc), "-> 선단 편성 계획 입력(검토 후 사용)")
# ---------------- 5) 운영 문서 완비율·인터페이스 명세 ----------------
docs = {"데이터 사전·스키마(M1)": 24, "정제 규칙 10종(M2)": 10, "Baseline·EDA(M3)": 8, "ETA 모델·피처(M4)": 21, "이탈·정차 경보(M5)": 3, "대시보드·설명(M6)": 4,
"승인 절차·권한(M7)": 4, "부하·오류 시험(M8)": 7, "AAR·ROI(M9)": 3, "운영·재학습 절차(M10)": 6, "장애 대응": 5, "인터페이스 가이드": 4}
done = sum(v > 0 for v in docs.values())
iface = {"input": {"telemetry": ["trip:int", "t:int", "x:float", "y:float", "speed:float", "heading:float", "rssi:float"], "format": "json-lines"},
"output": {"eta": ["trip", "eta_min", "plan_gap_min", "alarm", "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.trip >= 200)]
fin_mae, fin_mape = evaluate(model, te)
tt = te[te.eta_true >= 15]
base_mae, base_mape = mae(tt.eta_true, tt.base_eta), mape(tt.eta_true, tt.base_eta)
checks = [("환경 점검 통과", env_ok == len(libs)), ("데이터 정합성 전 항목 통과", all(chk.values())), ("재학습 파이프라인 정상 실행(챔피언/챌린저 판정 출력)", rt["champ_mae"] > 0),
("ETA MAPE 25% 이하", fin_mape <= 25), ("단순 ETA 대비 MAE 20% 이상 개선", fin_mae <= 0.8 * base_mae),
("7일 소요 예측이 전주 동요일 Baseline보다 정확", mape(dem[te_d], f_r) < mape(dem[te_d], f_n)),
("경보 건 자동 지시 0건(승인대기 분리)", bool((pend["처리"].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" - 신규 이동 40개를 포함해 재학습 판단을 자동 수행했다(드리프트 {'감지' if rt['drift'] else '미감지'}, {'챌린저 승격' if rt['promote'] else '기존 모델 유지'}).")
print(f" - 신규 구간 ETA MAPE {base_mape:.1f}%(단순 ETA) -> {fin_mape:.1f}%, MAE {base_mae:.1f}분 -> {fin_mae:.1f}분. 7일 납품 소요 예측 MAPE {mape(dem[te_d], f_r):.1f}%.")
print(f" - 120분 시점 선단 {len(live)}개 중 경보 {len(pend)}건은 승인대기로 분리했다. 실제 대체 수송은 수송통제관 결재와 지휘계통 결심에 따른다.")
fig, ax = plt.subplots(1, 2, figsize=(12, 4))
ax[0].scatter(tt.eta_true, tt.base_eta, s=4, alpha=.3, label="simple ETA"); ax[0].scatter(tt.eta_true, model.predict(tt[FEATS]), s=4, alpha=.3, label="model"); ax[0].plot([0, 300], [0, 300], "k--"); ax[0].legend(); ax[0].set_title("New trips: ETA actual vs predicted (min)")
ax[1].plot(days, dem, label="actual"); ax[1].plot(nxt, fc, "r--o", label="forecast"); ax[1].legend(); ax[1].set_title("Daily delivery transport demand and 7-day forecast")
plt.suptitle("MOV06 integrated pipeline (synthetic)"); plt.tight_layout(); plt.show()