Fashion Personalization · 행동 이벤트 기반 추천

해결한 문제와 담당 역할

같은 클릭이 두 번 들어가거나 처리 중 실패한 이벤트가 섞이면 사용자 취향 데이터가 틀어지고, 추천도 계속 엇나갑니다. 오래된 추천 결과를 최신인 줄 알고 보여 줄 수도 있습니다.

핵심 로직은 외부 서비스 없이 돌게 만들고, FastAPI는 필요할 때 붙이는 구조로 했습니다.

멱등 키로 중복을 막고, 재시도 한도와 DLQ, 재처리 흐름을 넣었습니다.

활성 사용자별 추천 스냅샷을 배치로 생성하고 생성 시각·반영 이벤트 수·미반영 상태를 기록했습니다.

핵심 요구사항

같은 멱등 키는 취향 데이터에 한 번만 반영됩니다.
세 번 실패한 이벤트는 DLQ로 빠집니다.
추천 결과에는 생성 시각과 반영한 이벤트 수가 같이 저장됩니다.
점수가 같으면 상품 ID 순서로 고정합니다.

Fashion Personalization Platform처리 흐름

01상품 목록
02사용자 행동 이벤트
03사용자 취향 데이터
04추천 순위
05미리 만든 추천 결과

행동 이벤트가 중복 확인과 재시도를 거쳐 취향 데이터, 추천 순위, 저장된 추천 결과로 이어지는 구성도. 주요 흐름은 상품 목록 → 사용자 행동 이벤트 → 사용자 취향 데이터 → 추천 순위 → 미리 만든 추천 결과 순서로 이어집니다.

기술 선택과 이유

핵심 로직은 프레임워크 없이

적용 방식: 추천과 이벤트 처리 로직은 FastAPI 없이도 돌아갑니다.
검토한 대안: FastAPI 모델과 ORM을 도메인 코드에서 바로 쓰기
선택 이유: 추천 규칙은 pytest만으로 빠르게 확인하고 싶었고, 나중에 HTTP 계층을 바꿔도 로직은 그대로 두고 싶었습니다.

실패한 이벤트는 따로

적용 방식: 중복 확인, 재시도, DLQ, 재처리를 단계별로 나눴습니다.
검토한 대안: 실패하면 될 때까지 바로 재시도
선택 이유: 계속 실패하는 이벤트 하나 때문에 다른 이벤트 처리가 막히면 안 됐습니다.

취향 갱신과 추천 생성은 따로

적용 방식: 이벤트가 올 때마다 취향 데이터를 바꾸고, 추천 결과는 배치로 따로 만듭니다.
검토한 대안: 추천 요청마다 이벤트를 전부 다시 집계
선택 이유: 스냅샷 조회와 추천 재계산을 구분하고 결과의 기준 시점을 추적하기 위한 선택입니다. 현재 HTTP 추천 API는 즉시 계산 경로를 사용합니다.

규칙 기반 점수

적용 방식: 브랜드, 카테고리, 태그, 가격, 사이즈, 재고 규칙으로 점수와 이유를 만들고, 점수가 같으면 상품 ID 순으로 정렬합니다.
검토한 대안: 머신러닝 모델이나 외부 추천 API 쓰기
선택 이유: 같은 입력이면 항상 같은 결과가 나와야 테스트할 수 있고, 왜 이 상품이 위에 있는지 코드로 설명할 수 있어야 했습니다.

주요 구현과 관련 테스트

핵심 로직과 동작을 확인한 테스트를 함께 정리했습니다.

주요 구현

`apply_event_to_profile`

src/fashion_personalization/recommendation.py

코드

소스 파일 열기

selected_size = event.payload.get("size")
if isinstance(selected_size, str) and selected_size in product.sizes:
    _add_weight(
        profile.size_preferences, selected_size, max(weight, 0.1)
    )

구현 내용

실제로 있는 사이즈 옵션을 골랐을 때만 사이즈 취향에 반영해서, 잘못된 값이 섞이지 않게 합니다.

`rank_products`

src/fashion_personalization/recommendation.py

코드

소스 파일 열기

score, reasons = _score_product(profile, product)
recommendations.append(
    Recommendation(product_id=product.product_id, score=round(score, 4), reasons=tuple(reasons))
)
return sorted(recommendations, key=lambda item: (-item.score, item.product_id))[:limit]

구현 내용

점수와 추천 이유를 같이 저장하고, 같은 점수는 상품 ID로 정렬합니다. 테스트는 사용자 취향이 상위 결과와 이유에 반영되는지 확인합니다.

`RecommendationBatchWorkflow.refresh_snapshots`

src/fashion_personalization/batch.py

코드

소스 파일 열기

snapshot = RecommendationSnapshot(
    generated_at=utc_now(),
    source_event_count=profile.event_count,
)
self.store.save_snapshot(snapshot)

구현 내용

추천 결과에 생성 시각과 반영 이벤트 수를 남겨서 얼마나 최신인지 알 수 있게 합니다.

테스트 결과

중복 처리 방지

같은 이벤트 2번

취향 데이터에는 1번만 반영

실패 이벤트 분리

처리 3번 실패

DLQ로 빠지고 나머지 이벤트는 계속 처리

Fashion Personalization Platform 테스트 결과
구분	테스트 방법	결과
단위 테스트	같은 멱등 키 이벤트와 세 번 실패하는 이벤트를 넣습니다.	중복은 한 번만 반영되고, 실패한 이벤트는 DLQ로 빠집니다.
단위 테스트	선호 브랜드, 사이즈, 가격, 재고 조건을 바꿔 가며 추천 순위를 확인합니다.	취향이 점수와 이유에 반영되고, 품절 상품은 빠지고, 이벤트 순서가 달라도 가격 취향은 같게 나옵니다.
통합 테스트	활성 사용자 대상으로 배치를 돌리고 최신 여부를 조회합니다.	고정 입력에서 스냅샷이 2개 이상 생성되고, 반영 이벤트 수가 프로필보다 적은 스냅샷은 0개입니다.

프로젝트 범위

검증 한계와 미구현 범위

저장소는 단일 프로세스의 메모리 구현입니다. 외부 브로커·Outbox·분산 저장소와 AWS 배포는 구현·검증하지 않았습니다.
POST /events는 이벤트를 대기 상태로 저장합니다. HTTP 이벤트 워커와 자동 배치 스케줄러는 없으며 서비스 메서드로 처리합니다.
추천은 정해 둔 규칙일 뿐 머신러닝이 아니고, 추천 품질을 평가하지도 않았습니다. 같은 점수일 때 ID 순서로 나오는지 직접 확인하는 테스트도 아직 없습니다.

향후 개선

PostgreSQL 어댑터를 붙이고 통합 테스트를 추가하려고 합니다.