이 글은 누구를 위한 것인가
- 재고 과잉과 품절이 반복되어 물류 비용과 기회 손실이 발생하고 있는 이커머스 운영팀
- ML 기반 수요 예측 시스템을 처음 도입하려는 데이터 엔지니어
- 자동 발주 시스템을 구축하여 바이어의 업무를 자동화하고 싶은 개발자
들어가며
재고 관리는 이커머스에서 가장 어려운 문제 중 하나다. 너무 많이 쌓으면 보관 비용이 늘고, 너무 적으면 품절로 매출 기회를 잃는다. 전통적인 방식은 바이어가 과거 판매 데이터와 직감으로 발주량을 결정하는 것이었다.
ML 기반 수요 예측은 이 문제를 구조적으로 해결한다. 계절성, 프로모션 효과, 경쟁사 재고 상황, 날씨까지 고려한 예측이 가능하다. 실제로 ML 수요 예측을 도입한 기업들은 재고 비용을 평균 30% 줄이고, 품절률을 50% 이상 낮추는 효과를 보고 있다.
이 글은 bluefoxdev.kr의 이커머스 데이터 파이프라인 구축 을 참고하고, 재고 최적화 관점에서 확장하여 작성했습니다.
1. 수요 예측 모델 선택
1.1 모델 비교
| 모델 | 장점 | 단점 | 적합한 경우 |
|---|---|---|---|
| Prophet | 구현 쉬움, 해석 가능 | 복잡한 패턴 한계 | 초기 도입, 명확한 계절성 |
| LSTM | 복잡한 패턴 학습 | 데이터 많이 필요 | 대규모 SKU, 충분한 히스토리 |
| LightGBM | 빠른 학습, 정확도 높음 | 특성 공학 필요 | 다변수 예측 |
| ARIMA | 통계적 해석 가능 | 자동화 어려움 | 단순한 시계열 |
초기에는 Prophet, 데이터가 쌓이면 LightGBM 앙상블로 발전시키는 것이 현실적이다.
1.2 Prophet 기반 수요 예측 구현
from prophet import Prophet
import pandas as pd
import numpy as np
class DemandForecastService:
def __init__(self, product_id: str):
self.product_id = product_id
self.model = Prophet(
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False,
seasonality_mode='multiplicative' # 곱셈형: 성수기 효과 반영
)
# 한국 공휴일 추가
self.model.add_country_holidays(country_name='KR')
def prepare_data(self, sales_df: pd.DataFrame) -> pd.DataFrame:
"""Prophet 형식으로 데이터 변환"""
return sales_df.rename(columns={
'date': 'ds',
'quantity_sold': 'y'
})[['ds', 'y']]
def add_promotions(self, promotions: list[dict]):
"""프로모션 이벤트를 외생 변수로 추가"""
for promo in promotions:
self.model.add_regressor(promo['name'])
def train(self, sales_df: pd.DataFrame):
df = self.prepare_data(sales_df)
self.model.fit(df)
def forecast(self, days: int = 30) -> pd.DataFrame:
future = self.model.make_future_dataframe(periods=days)
forecast = self.model.predict(future)
return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(days)
2. 안전 재고 계산
2.1 안전 재고 공식
안전 재고는 예측 불확실성과 리드타임을 고려한 버퍼다.
안전 재고 = Z × σ_수요 × √리드타임
Z: 서비스 수준에 따른 안전 계수
- 95% 서비스 수준: Z = 1.65
- 99% 서비스 수준: Z = 2.33
σ_수요: 수요의 표준편차
리드타임: 발주 후 입고까지 기간 (일)
import numpy as np
from scipy import stats
def calculate_safety_stock(
demand_history: list[float],
lead_time_days: int,
service_level: float = 0.95
) -> int:
"""
service_level: 0.95 = 95% 서비스 수준
"""
z_score = stats.norm.ppf(service_level) # 1.645
demand_std = np.std(demand_history)
safety_stock = z_score * demand_std * np.sqrt(lead_time_days)
return int(np.ceil(safety_stock))
def calculate_reorder_point(
avg_daily_demand: float,
lead_time_days: int,
safety_stock: int
) -> int:
"""발주 시점 재고 수준 계산"""
return int(avg_daily_demand * lead_time_days + safety_stock)
3. 자동 발주 시스템 설계
3.1 발주 트리거 로직
[재고 모니터링 스케줄러 - 매일 오전 9시]
↓
현재 재고 조회
↓
재주문점(ROP) 이하인 상품 탐지
↓
수요 예측 기반 발주량 계산
↓
최소 발주 단위(MOQ) 적용
↓
발주 초안 생성
↓
바이어 검토 (고가/신규 상품) 또는 자동 발주
3.2 발주량 최적화
def calculate_optimal_order_quantity(
forecast_demand: float, # 리드타임 기간 예측 수요
safety_stock: int, # 안전 재고
current_stock: int, # 현재 재고
target_days_of_stock: int, # 목표 재고 보유 일수
avg_daily_demand: float, # 일 평균 수요
moq: int = 1 # 최소 발주 단위
) -> int:
target_stock = avg_daily_demand * target_days_of_stock + safety_stock
order_quantity = max(0, target_stock - current_stock)
# MOQ 반올림
if order_quantity > 0:
order_quantity = max(moq, round(order_quantity / moq) * moq)
return order_quantity
class AutoReorderService:
async def run_daily_check(self):
products = await self.get_active_products()
for product in products:
current_stock = await self.inventory.get_current_stock(product.id)
rop = await self.get_reorder_point(product.id)
if current_stock <= rop:
order_qty = calculate_optimal_order_quantity(
forecast_demand=await self.get_forecast(product.id, product.lead_time_days),
safety_stock=product.safety_stock,
current_stock=current_stock,
target_days_of_stock=45,
avg_daily_demand=product.avg_daily_demand,
moq=product.minimum_order_quantity
)
if order_qty > 0:
await self.create_purchase_order(product, order_qty)
4. 계절성과 프로모션 보정
4.1 주요 보정 요소
수요 예측 × 보정 계수
계절성 보정:
- 여름 성수기 (6~8월): 1.3~1.8배
- 겨울 성수기 (11~1월): 1.4~2.0배
이벤트 보정:
- 블랙프라이데이: 3~5배
- 설날/추석: 1.5~2.5배
- 여름 세일: 2~4배
신상품 보정:
- 출시 첫 달: 보수적 예측 (히스토리 없음)
- 유사 상품 히스토리 활용
4.2 예측 정확도 모니터링
def calculate_forecast_accuracy(
actual: list[float],
predicted: list[float]
) -> dict:
"""
MAPE: Mean Absolute Percentage Error
RMSE: Root Mean Square Error
"""
actual = np.array(actual)
predicted = np.array(predicted)
mape = np.mean(np.abs((actual - predicted) / actual)) * 100
rmse = np.sqrt(np.mean((actual - predicted) ** 2))
return {
'mape': round(mape, 2),
'rmse': round(rmse, 2),
'accuracy_pct': round(100 - mape, 2)
}
MAPE 15% 이하면 양호, 10% 이하면 우수 수준이다.
마무리: 도입 로드맵
| 단계 | 기간 | 목표 |
|---|---|---|
| 1단계 | 1개월 | 데이터 파이프라인 구축, 히스토리 정리 |
| 2단계 | 2개월 | Prophet 모델 적용, 예측 정확도 측정 |
| 3단계 | 3개월 | 안전 재고 자동 계산, 발주 알림 자동화 |
| 4단계 | 6개월+ | 완전 자동 발주, 앙상블 모델 고도화 |
ML 수요 예측은 완벽하지 않다. 하지만 사람의 직감보다 일관성 있고 확장 가능하다. 예측 정확도를 지속적으로 모니터링하고 모델을 개선하는 것이 성공의 핵심이다.