♻️ [Refactor/#40] 계산엔진점검

.gitignore

@@ -63,6 +63,11 @@ htmlcov/
 *.sqlite3
 mysql_data/
 data/
+# 런타임에 필요한 집계 통계 파일만 예외적으로 커밋한다(개인 단위 원본 데이터 아님).
+# 디렉터리를 먼저 풀어준 뒤 그 안의 나머지 파일은 다시 전부 무시하고, 대상 파일 하나만 재포함한다.
+!fastapi/data/
+fastapi/data/*
+!fastapi/data/active_income_stats.csv
 
 # =========================
 # Docker

fastapi/app/error_handlers.py

@@ -0,0 +1,78 @@
+"""전 엔드포인트의 에러 응답을 app/schemas/errors.py의 단일 포맷으로 통일하는 핸들러.
+
+- RequestValidationError(Pydantic 422): 필드별 원인을 한국어로 번역해 details에 담는다.
+- HTTPException(라우터가 ValueError를 잡아 400/500으로 던진 것): 동일 포맷으로 감싼다.
+"""
+
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+from app.schemas.errors import ErrorDetail, ErrorResponse
+
+_PYDANTIC_ERROR_MESSAGES: dict[str, str] = {
+    "missing": "필수 항목입니다.",
+    "int_parsing": "정수여야 합니다.",
+    "int_type": "정수여야 합니다.",
+    "float_parsing": "숫자여야 합니다.",
+    "float_type": "숫자여야 합니다.",
+    "string_type": "문자열이어야 합니다.",
+    "enum": "허용되지 않는 값입니다.",
+    "bool_parsing": "true/false 값이어야 합니다.",
+    "bool_type": "true/false 값이어야 합니다.",
+}
+
+
+def _translate_pydantic_error(error: dict) -> str:
+    """Pydantic 에러 하나를 한국어 문구로 번역한다. 매핑에 없는 타입은 원문 메시지를 그대로 쓴다."""
+    error_type = error.get("type", "")
+    if error_type == "value_error":
+        # model_validator가 던진 ValueError는 이미 한국어지만, Pydantic이 "Value error, " 접두사를 붙인다.
+        return error.get("msg", "").removeprefix("Value error, ")
+    return _PYDANTIC_ERROR_MESSAGES.get(error_type, error.get("msg", "입력값이 올바르지 않습니다."))
+
+
+def _field_name(loc: tuple) -> str | None:
+    """Pydantic의 loc(예: ("body", "current_age"))를 "current_age" 같은 점 표기 필드명으로 변환한다.
+    특정 필드에 속하지 않는 에러(예: 모델 전체에 대한 검증)면 None을 반환한다."""
+    path = [str(part) for part in loc if part != "body"]
+    return ".".join(path) if path else None
+
+
+def _validation_error_message(details: list[ErrorDetail]) -> str:
+    if len(details) == 1:
+        return details[0].reason
+    return f"입력값을 확인해주세요. ({len(details)}개 항목에 문제가 있습니다.)"
+
+
+_STATUS_CODE_TO_ERROR_CODE: dict[int, str] = {
+    400: "INVALID_INPUT",
+    404: "NOT_FOUND",
+    500: "INTERNAL_ERROR",
+}
+
+
+def _error_code_for_status(status_code: int) -> str:
+    return _STATUS_CODE_TO_ERROR_CODE.get(status_code, "ERROR")
+
+
+def register_error_handlers(app: FastAPI) -> None:
+    @app.exception_handler(RequestValidationError)
+    async def _handle_validation_error(request: Request, exc: RequestValidationError) -> JSONResponse:
+        details = [
+            ErrorDetail(field=_field_name(error["loc"]), reason=_translate_pydantic_error(error))
+            for error in exc.errors()
+        ]
+        body = ErrorResponse(
+            code="VALIDATION_ERROR",
+            message=_validation_error_message(details),
+            details=details,
+        )
+        return JSONResponse(status_code=422, content=body.model_dump())
+
+    @app.exception_handler(StarletteHTTPException)
+    async def _handle_http_exception(request: Request, exc: StarletteHTTPException) -> JSONResponse:
+        message = exc.detail if isinstance(exc.detail, str) else "요청을 처리할 수 없습니다."
+        body = ErrorResponse(code=_error_code_for_status(exc.status_code), message=message, details=[])
+        return JSONResponse(status_code=exc.status_code, content=body.model_dump())

fastapi/app/main.py

@@ -1,14 +1,28 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 
+from app.error_handlers import register_error_handlers
+from app.repositories.employees_income_repository import ensure_data_available
 from app.routers import employees, retirement
 
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    ensure_data_available()
+    yield
+
+
 app = FastAPI(
     title="ORBIT FastAPI Server",
     description="ORBIT FastAPI Server",
     version="0.1.0",
     root_path="/ai",
+    lifespan=lifespan,
 )
 
+register_error_handlers(app)
+
 app.include_router(employees.router)
 app.include_router(retirement.router)
 

fastapi/app/repositories/employees_income_repository.py

@@ -6,6 +6,19 @@
 _stats: pd.DataFrame | None = None
 
 
+def ensure_data_available() -> None:
+    """/api/employees/simulate에 필요한 소득 통계 CSV의 존재를 앱 기동 시점에 검증한다.
+
+    파일이 없으면 첫 요청에서 500이 나는 대신, 여기서 즉시 RuntimeError로 기동을 실패시킨다.
+    """
+    if not _CSV_PATH.exists():
+        raise RuntimeError(
+            f"필수 데이터 파일이 없습니다: {_CSV_PATH}\n"
+            "재직자 연금 시뮬레이션(/api/employees/simulate)에 필요한 소득 통계 파일(active_income_stats.csv)입니다. "
+            "scripts/preprocess.py로 생성하거나, 배포 환경에서 데이터 볼륨이 올바르게 마운트됐는지 확인하세요."
+        )
+
+
 def _get_stats() -> pd.DataFrame:
     global _stats
     if _stats is None:

fastapi/app/schemas/errors.py

@@ -0,0 +1,20 @@
+"""전 엔드포인트가 공유하는 단일 에러 응답 스키마.
+
+Pydantic 422(RequestValidationError)와 라우터의 HTTPException(400/500 등)을
+app/main.py의 예외 핸들러가 전부 이 형태로 변환해서 내보낸다. 프론트는 항상
+code로 분기하고, message를 사용자에게 그대로 보여주고, details로 어떤 필드가
+문제인지 알 수 있다.
+"""
+
+from pydantic import BaseModel
+
+
+class ErrorDetail(BaseModel):
+    field: str | None  # 문제가 된 필드명(점 표기, 예: "current_age"). 특정 필드에 속하지 않으면 None
+    reason: str  # 해당 필드가 왜 문제인지에 대한 한국어 설명
+
+
+class ErrorResponse(BaseModel):
+    code: str  # 프론트 분기용 에러 코드 (예: VALIDATION_ERROR, INVALID_INPUT, INTERNAL_ERROR)
+    message: str  # 사용자에게 그대로 보여줄 수 있는 한국어 메시지
+    details: list[ErrorDetail] = []  # 필드별 상세 (없으면 빈 리스트)

fastapi/app/schemas/money.py

@@ -0,0 +1,30 @@
+"""API 경계의 금액 단위(원) <-> 내부 계산 단위(만원) 변환을 한 곳에 모은 모듈.
+
+app/services/* 의 계산 로직과 상수(INVESTMENT_RETURN, reduction_rules의 threshold 등)는
+전부 만원 단위를 그대로 사용한다(변경하지 않음). 이 모듈이 제공하는 타입 별칭을
+요청/응답 스키마의 금액 필드에 붙이면, Pydantic이 역직렬화/직렬화 시점에 자동으로
+원 <-> 만원 변환을 수행한다. 다른 곳에서 별도로 곱셈/나눗셈을 하지 않는다.
+"""
+
+from typing import Annotated
+
+from pydantic import BeforeValidator, PlainSerializer
+
+WON_PER_MANWON = 10_000
+
+
+def won_to_manwon(value: float) -> float:
+    """API 요청으로 들어온 원 단위 금액을 내부 계산용 만원 단위로 변환한다."""
+    return value / WON_PER_MANWON
+
+
+def manwon_to_won(value: float) -> int:
+    """내부 계산 결과(만원)를 API 응답용 원 단위 정수로 변환한다."""
+    return round(value * WON_PER_MANWON)
+
+
+WonAmountInput = Annotated[float, BeforeValidator(won_to_manwon)]
+"""요청 스키마의 금액 필드에 사용한다. API 입력은 원 단위, 필드에는 만원 단위로 저장된다."""
+
+WonAmountOutput = Annotated[float, PlainSerializer(manwon_to_won, return_type=int)]
+"""응답 스키마의 금액 필드에 사용한다. 내부(만원) 값을 API 출력 시 원 단위 정수로 직렬화한다."""

fastapi/app/schemas/retirement.py

@@ -2,6 +2,15 @@
 
 from pydantic import BaseModel
 
+from app.schemas.money import WonAmountInput, WonAmountOutput
+
+
+class Gender(str, Enum):
+    """성별. target_age(목표연령) 산정에 사용된다."""
+
+    MALE = "male"
+    FEMALE = "female"
+
 
 class ReadinessStatus(str, Enum):
     """노후 준비 상태 판정 결과"""
@@ -13,28 +22,29 @@ class ReadinessStatus(str, Enum):
 
 class TimelinePoint(BaseModel):
     age: int  # 나이
-    asset: float  # 해당 나이 시점의 자산
-    income: float  # 해당 연도 소득(연금)
-    expense: float  # 해당 연도 지출
-    gap: float  # 연간 Gap (지출 - 소득)
-    cumulative_gap: float  # 누적 Gap
+    asset: WonAmountOutput  # 해당 나이 시점의 자산 (원)
+    annual_income: WonAmountOutput  # 해당 연도 소득(연금) (원, 연 단위)
+    annual_expense: WonAmountOutput  # 해당 연도 지출 (원, 연 단위)
+    annual_gap: WonAmountOutput  # 연간 Gap (지출 - 소득) (원, 연 단위)
+    cumulative_annual_gap: WonAmountOutput  # 누적 Gap (원, annual_gap의 누적합)
 
 
 class SimulationResult(BaseModel):
     current_age: int  # 현재 나이
-    monthly_gap: float  # 현재 시점의 월 Gap (월 생활비 - 월 연금)
+    monthly_gap: WonAmountOutput  # 현재 시점의 월 Gap (월 생활비 - 월 연금) (원)
     depletion_age: int | None  # 자산 고갈 나이 (고갈되지 않으면 None)
+    depleted: bool  # 자산 고갈 여부 (depletion_age is not None과 항상 일치)
     target_age: int  # 목표연령 (성별 고정값)
     status: ReadinessStatus  # 노후 준비 상태
     timeline: list[TimelinePoint]  # 연도별 자산 추이
 
 
 class DiagnosisRequest(BaseModel):
     current_age: int  # 현재 나이
-    monthly_expenses: float  # 월 생활비
-    monthly_pension: float  # 월 연금 수령액
-    asset: float  # 현재 보유 자산
-    gender: str  # 성별 ("male" 또는 "female")
+    monthly_expenses: WonAmountInput  # 월 생활비 (원)
+    monthly_pension: WonAmountInput  # 월 연금 수령액 (원)
+    asset: WonAmountInput  # 현재 보유 자산 (원)
+    gender: Gender  # 성별
 
 
 class RecommendationRequest(DiagnosisRequest):
@@ -52,10 +62,11 @@ class RecommendationType(str, Enum):
 class RecommendationResult(BaseModel):
     current_age: int  # 현재 나이
     recommendation_type: RecommendationType  # 추천 유형
-    required_saving: float  # 필요 월 절약액 (만원)
-    required_income: float  # 필요 월 추가 소득액 (만원)
+    required_saving: WonAmountOutput  # 필요 월 절약액 (원)
+    required_income: WonAmountOutput  # 필요 월 추가 소득액 (원)
     target_status: ReadinessStatus  # 추천 산정에 사용된 목표 기준 (항상 SUFFICIENT="고갈 없음")
     depletion_age: int | None  # 개선 적용 후 자산 고갈 나이
+    depleted: bool  # 개선 적용 후 자산 고갈 여부 (depletion_age is not None과 항상 일치)
     target_age: int  # 목표연령
     status: ReadinessStatus  # 개선 적용 후 노후 준비 상태
     timeline: list[TimelinePoint]  # 개선 적용 후 연도별 자산 추이
@@ -64,17 +75,18 @@ class RecommendationResult(BaseModel):
 class ReductionRequest(DiagnosisRequest):
     """진단(diagnosis)과 동일한 입력 스키마를 재사용하고 재취업 관련 필드를 추가"""
 
-    reemployment_income: float  # 재취업 예상 월소득 (만원)
+    reemployment_income: WonAmountInput  # 재취업 예상 월소득 (원)
     year: int | None = None  # 소득심사 기준 연도 (미지정 시 최신 규칙 적용)
 
 
 class ReductionResult(BaseModel):
     current_age: int  # 현재 나이
-    reemployment_income: float  # 재취업 예상 월소득 (만원)
-    monthly_reduction: float  # 월 감액액 (만원)
-    reduced_monthly_pension: float  # 감액 후 월 실수령 연금액 (만원)
-    full_payment_income_threshold: float  # 전액 수령 가능 소득 상한 (만원)
+    reemployment_income: WonAmountOutput  # 재취업 예상 월소득 (원)
+    monthly_reduction: WonAmountOutput  # 월 감액액 (원)
+    reduced_monthly_pension: WonAmountOutput  # 감액 후 월 실수령 연금액 (원)
+    full_payment_income_threshold: WonAmountOutput  # 전액 수령 가능 소득 상한 (원)
     depletion_age: int | None  # 감액 반영 후 자산 고갈 나이
+    depleted: bool  # 감액 반영 후 자산 고갈 여부 (depletion_age is not None과 항상 일치)
     target_age: int  # 목표연령
     status: ReadinessStatus  # 감액 반영 후 노후 준비 상태
     timeline: list[TimelinePoint]  # 감액 반영 후 연도별 자산 추이
@@ -84,7 +96,7 @@ class ScenariosRequest(DiagnosisRequest):
     """진단(diagnosis)과 동일한 입력 스키마를 재사용하고 수령방식 계산에 필요한 필드를 추가"""
 
     early_years: int = 5  # 조기수령 연수 (1~5년, 1년당 5% 감액)
-    base_monthly_income: float  # 기준소득월액 (만원) - LUMP_SUM/SPLIT 공제일시금 산식에 사용
+    base_monthly_income: WonAmountInput  # 기준소득월액 (원) - LUMP_SUM/SPLIT 공제일시금 산식에 사용
     total_service_years: int  # 총 재직연수 - LUMP_SUM/SPLIT 공제일시금 산식에 사용
     deduction_years: int | None = None  # SPLIT(분할수령) 공제연수. 미지정 시 제약 내 최댓값으로 클램프
 
@@ -100,8 +112,9 @@ class ScenarioType(str, Enum):
 
 class ScenarioOutcome(BaseModel):
     scenario_type: ScenarioType  # 수령방식
-    depletion_age: int  # 자산 고갈 나이 (고갈되지 않으면 MAX_AGE)
-    total_received: float  # 총 수령액 (만원, current_age~MAX_AGE 동일 기간 기준)
+    depletion_age: int | None  # 자산 고갈 나이 (고갈되지 않으면 None)
+    depleted: bool  # 자산 고갈 여부 (depletion_age is not None과 항상 일치)
+    total_received: WonAmountOutput  # 총 수령액 (원, current_age~MAX_AGE 동일 기간 기준)
     break_even_age: int | None  # NORMAL 대비 손익분기 나이 (이 나이 이상 생존 시 NORMAL이 유리해짐). NORMAL 자신은 None
     timeline: list[TimelinePoint]  # 연도별 자산 추이