교통 흐름 모델링과 자율주행 데이터 운영 설계

Traffic Flow Modeling과 Autonomous Driving Data를 통합해 교통 예측·제어·검증 파이프라인을 설계하는 데이터 거버넌스와 운영 원칙을 정리한다.

2026-08-14 · 최초 발행 2025-10-14

예측·제어·검증을 하나의 데이터 루프로 묶기

도시 혼잡을 완화하고 자율주행의 안전성을 높이려면 교통 흐름과 차량 운행 데이터를 따로 다룰 수 없다. 대규모 시계열·공간 데이터, 시뮬레이션, 실차 운행 로그를 연결해 예측·제어·검증이 순환하는 체계가 필요하다.

Traffic Flow Modeling은 도로 네트워크에서 차량 흐름을 수리·통계·시뮬레이션으로 표현하는 방식이다. 연속류를 보는 마크로스코픽 모델부터 메조스코픽 모델, 개별 차량을 다루는 마이크로스코픽 모델까지 범위가 넓다. 이동시간과 큐 길이를 줄이거나 throughput을 높이고, 충돌위험·급제동 이벤트 같은 안전지표를 개선하는 목적 함수를 둘 수 있다.

Autonomous Driving Data는 라이다·레이다·카메라·GNSS/IMU 같은 멀티모달 센서, HD Map, V2X, 차량 CAN 신호, 시나리오 메타데이터로 구성된다. 원시 로그는 ROSbag/MCAP, 라벨은 OpenLABEL, 시나리오는 OpenSCENARIO/OpenDRIVE 형식을 사용할 수 있으며 캘리브레이션 정보와 모델 아티팩트도 포함된다. 표준은 진화 중이므로 최신 정보 확인이 필요하다.

이 둘은 데이터 수집, 정합성 검증, 라벨링·QA, 특징 생성, 학습·튜닝, 시뮬레이션 검증, 배포, 모니터링, 하드케이스 채집, 재학습으로 이어지는 폐루프로 결합된다. 대용량 학습은 오프라인 배치로, 실시간 제어와 예측은 온라인 스트리밍으로 분리해 운영한다.

데이터 품질이 모델보다 먼저 흔들리는 지점

도로 인프라의 지능형 신호·검지기, 플릿 센서, 날씨·공사·이벤트 같은 외부 데이터를 통합할 때는 링크ID, 타임스탬프, 좌표계를 공통 키로 표준화해야 한다. 스키마 버저닝과 캘리브레이션 이력, 라벨 정확도·일치도·재현성 지표를 관리하고, 개인정보와 차량식별 정보에는 비식별화 및 접근통제 정책을 적용한다.

모델 선택에는 뚜렷한 교환 조건이 있다. CTM, IDM, Gipps 같은 물리·규칙 기반 모델은 해석 가능성과 안정성에 강점이 있지만 복잡한 상황에 적응하는 데 한계가 있다. GNN·LSTM·Transformer, RL/MPC 하이브리드처럼 데이터 기반 접근은 예측력과 적응성이 좋지만 데이터 품질 의존도와 운영 복잡도가 높아진다. 이 경우 XAI와 안전 제약으로 해석 가능성을 보완할 필요가 있다.

파이프라인은 Ingest에서 시작해 Validate, Label/Auto-label, Feature Store, Train/HP Tuning, Scenario-based Sim, AB/Shadow Deploy, Monitor, Active Learning으로 이어진다. 데이터 레이크에는 ACID와 타임트래블을, 모델 레지스트리에는 스테이지 전이와 승인 워크플로를 적용할 수 있다. 실패 격리와 재시도, 아이들 포크 방지까지 고려한 내고장 설계가 요구된다.

실시간 경로에서는 카프카/파ulsar 기반 스트리밍 피처 생성, ONNX/TensorRT 저지연 추론, 캐시 전략을 함께 구성한다. ingest+feature+infer 지연 예산은 300~800ms가 목표다. 신호제어와 램프미터링에 MPC/RL을 적용할 때는 최소 녹색시간과 보행자 보호 같은 안전 제약 아래에서 제어를 최적화해야 한다.

전송·저장 암호화, 키 관리, 플릿 단말의 보안부팅과 원격 업데이트 검증, 데이터 보존·삭제 주기도 운영 범위에 들어간다. 테스트와 트레인을 분리하고 시나리오 다양성 커버리지와 ODD(운행영역)를 명시해 모니터링한다. ASAM 계열 표준과 지역 규제 준수 여부도 최신 정보 확인이 필요하다.

입력 데이터부터 모델 승인까지의 흐름

DeployProcessingStorageIngestEdgeHard casesSchema mismatchCalib failVersioningVehicle SensorsLiDAR/Radar/Camera/GNSS-IMUROSbag/MCAP LogsSchema & Time SyncValidation(OpenLABEL/ASAM)Calibration ConsistencyCheckData Lake (Delta/Hudi)ACID + Time TravelMetadata Catalog+ Model RegistryAuto-Label + QA(Human-in-the-loop)Feature Store(Batch/Streaming)Training & Tuning(GNN/LSTM/RL)Scenario Simulation(SUMO/CARLA)Serving: ONNX/TensorRTA/B, ShadowMonitoring: Drift & SafetyKPIsQuarantine + AlertDead-letter QueueRecall/RecalibrateBlock for Training

이 구조에서는 일관성 관리가 핵심이다. Delta/Hudi의 스냅샷 격리와 커밋 로그 기반 멀티라이터 제어, 피처 스토어의 오프셋·이벤트타임 정합성을 보장해야 한다. 스키마 불일치는 격리 큐로 보내고, 캘리브레이션 실패는 학습을 차단하며, 재처리 워크플로는 자동화한다.

모델 레지스트리에서 Staging→Production 단계로 전이할 때는 인간 검토와 시나리오 커버리지 리포트를 승인 조건으로 둔다.

교통 모델은 해상도와 운영 복잡도를 함께 선택한다

접근법 성능 확장성 일관성 안정성 운영 편의
마크로스코픽(연속류) 혼잡 추세·링크 단위 예측 강점 네트워크 대규모 확장 용이 파라미터 안정, 변동성 낮음 외란에 안정적 모델 구축·튜닝 용이
메조스코픽 수요·공급 상호작용 표현 균형 중간 규모 네트워크 적합 합리적 일관성 시뮬 시간/정밀도 균형 운영 복잡도 중간
마이크로스코픽(개별차량) 세밀한 상호작용·안전 지표 우수 대규모 시뮬에 비용 증가 파라미터 민감도 높음 초기 조건 영향 큼 시나리오 관리 부담
데이터 기반(딥러닝/GNN) 고성능 예측·적응성 우수 분산 학습로 수평 확장 데이터 품질 의존도 큼 드리프트 리스크 존재 MLOps 요구 수준 높음
하이브리드(MPC+RL) 제약 하 최적 제어 성능 계산 비용 관리 필요 제약으로 일관성 확보 튜닝 복잡 운영·검증 복합도 높음

제어 현장과 차량 플릿에서의 적용

도시 신호 최적화에는 검지기·카메라 트래픽 시계열과 행사·공사 이벤트 메타데이터가 입력된다. LSTM/Transformer로 수요를 예측하고 MPC/RL 기반으로 신호군을 최적화한 뒤 보행·버스우선 제약을 적용한다. 결과는 주기·스플릿·오프셋 자동 조정과 평균 지체·큐 길이 KPI의 실시간 모니터링으로 이어진다.

고속도로 램프 미터링과 차로 제어는 루프 검지기·레이더 속도 및 점유율, 사고 감지 이벤트를 사용한다. CTM으로 임계점을 추정하고 짧은 수평선 예측과 시나리오 기반 안전 검증을 결합한다. 유입률과 가변속도를 설정해 쇼크웨이브 완화를 목표로 하며, 도로·수요에 따라 15~30% 충돌 위험 지표 개선을 기대할 수 있다.

자율주행 인지·예측 데이터 큐레이션에서는 라이다·카메라 원시 로그, 지도 레이어, 차량 행동 라벨을 다룬다. 오토라벨링과 휴먼검증, 클래스 불균형 재샘플링, 하드케이스 마이닝을 거쳐 버전 관리된 학습 세트와 시나리오 라이브러리, 재현 가능한 실험 메타데이터를 만든다.

플릿 러닝과 Shadow 모드 검증은 프로덕션 추론 로그, 성능·안전 KPI, 드리프트 지표를 입력으로 사용한다. Shadow 배포, A/B 테스트, 성능 회귀 테스트를 수행한 뒤 승인된 모델 전이, 롤백 전략, 규제 대응 리포트를 남긴다.

LSTM으로 링크 속도 시계열을 예측하는 예시

예시는 Python 3.10+, PyTorch 2.3+ 환경에서 CPU 실행을 전제로 하며, 합성 시계열의 입력 12 스텝으로 출력 3 스텝을 예측한다.

# pip install torch==2.3.0 numpy scikit-learn
import math, random, numpy as np
import torch
from torch import nn
from sklearn.metrics import mean_absolute_percentage_error

torch.manual_seed(42); np.random.seed(42); random.seed(42)
device = torch.device("cpu")

# 1) 데이터 생성: 일 변동 + 주간 변동 + 잡음
T = 2000
t = np.arange(T)
signal = 60 + 15*np.sin(2*math.pi*t/96) + 5*np.sin(2*math.pi*t/672) + np.random.normal(0,1.5,T)
signal = np.clip(signal, 0, 120)  # 속도 km/h

def make_seq(arr, in_len=12, out_len=3):
    X, y = [], []
    for i in range(len(arr)-in_len-out_len):
        X.append(arr[i:i+in_len])
        y.append(arr[i+in_len:i+in_len+out_len])
    return np.array(X)[:, :, None], np.array(y)  # (N, L, 1), (N, O)

X, y = make_seq(signal, 12, 3)
n_train = int(len(X)*0.8)
Xtr, Xte = X[:n_train], X[n_train:]
ytr, yte = y[:n_train], y[n_train:]

Xtr = torch.tensor(Xtr, dtype=torch.float32).to(device)
ytr = torch.tensor(ytr, dtype=torch.float32).to(device)
Xte = torch.tensor(Xte, dtype=torch.float32).to(device)
yte = torch.tensor(yte, dtype=torch.float32).to(device)

# 2) 모델 정의
class LSTMForecaster(nn.Module):
    def __init__(self, input_size=1, hidden=64, layers=2, out_len=3):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden, num_layers=layers, batch_first=True)
        self.head = nn.Sequential(nn.Linear(hidden, 64), nn.ReLU(), nn.Linear(64, out_len))
    def forward(self, x):
        # x: (B, L, 1)
        out, _ = self.lstm(x)
        h = out[:, -1, :]
        return self.head(h)

model = LSTMForecaster().to(device)
opt = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5)
loss_fn = nn.L1Loss()

# 3) 학습 루프
for epoch in range(30):
    model.train()
    perm = torch.randperm(Xtr.size(0))
    for i in range(0, Xtr.size(0), 128):
        idx = perm[i:i+128]
        xb, yb = Xtr[idx], ytr[idx]
        pred = model(xb)
        loss = loss_fn(pred, yb)
        opt.zero_grad(); loss.backward(); opt.step()
    if (epoch+1) % 10 == 0:
        model.eval()
        with torch.no_grad():
            te_pred = model(Xte).cpu().numpy()
            mape = mean_absolute_percentage_error(yte.cpu().numpy().ravel(), te_pred.ravel())
        print(f"epoch {epoch+1:02d} mape={mape*100:.2f}%")

# 4) 추론 예시
model.eval()
with torch.no_grad():
    sample = Xte[:1]
    forecast = model(sample).cpu().numpy().flatten()
print("next 3-step forecast (km/h):", np.round(forecast, 2))

실데이터에 적용할 때는 링크별·구간별 모델 또는 멀티링크 GNN 확장을 검토한다. 결측 처리, 날씨·사고 이벤트 특성 조인, 이벤트타임 기반 윈도잉이 필요하다. 배포 단계에서는 ONNX 변환, TensorRT 저지연 런타임 추론, 피처 스토어 오프셋 정합성을 관리한다.

운영 성과를 측정할 기준

도시 구조와 집행 수준에 따라 평균 지체·여행시간은 1025%, 큐 길이는 1530% 감소할 가능성이 있다. 급제동·근접 이벤트로 보는 사고 위험 지표는 1030% 개선되고, 신호 무정차 통행률은 515% 향상될 수 있다.

현장 조사와 수동 튜닝 시간은 30~50% 절감할 수 있으며, 모델 실험·검증 자동화는 리드타임 단축으로 이어진다. 데이터 중심 의사결정 문화를 정착시키고 시나리오 기반 안전 검증 체계를 마련하며, 규제 대응·감사 추적성과 시민 체감 서비스 품질을 강화하는 기반이 된다.

초기에는 마크로·하이브리드 접근으로 안정성을 확보하고, 이후 데이터 기반 모델을 점진적으로 도입하는 전략이 적합하다.

교통 흐름 모델링자율주행 데이터MLOps시뮬레이션신호 최적화