d01 - 7. 분석 로직 구현

7교시 · 분석 로직 구현

1일차 17:00–18:00 · 이론/실습 선행: 6교시(CsvSalesRepository), 4교시(엔티티·포트)

---

🎯 학습 목표

• 매출 지표를 계산하는 유스케이스를 TDD로 구현한다.
• 유스케이스가 포트(SalesRepository)만 의존하고, 외부(pandas·파일)는 모르게 만든다.
• Fake 리포지토리로 파일 없이 분석 로직만 빠르게 검증한다.

---

📖 개념 1 — 유스케이스는 "흐름의 조율자"

유스케이스(2계층)는 "이 앱이 사용자에게 제공하는 하나의 기능"을 담당합니다. 여기서는 "매출 레코드 목록을 받아 분석 결과를 만든다" 가 그 기능입니다.

중요한 것은 유스케이스가 무엇을 모르는가 입니다.

• 데이터가 어디서 왔는지 모릅니다 (CSV인지 DB인지). → 포트로부터 받을 뿐.
• pandas를 모릅니다. → 순수 파이썬으로 계산.
• 화면을 모릅니다. → 결과 객체만 돌려줄 뿐.

이렇게 "모르는 것이 많은" 덕분에, 유스케이스는 어떤 환경에서도 그대로 재사용되고, 외부 도구 없이 테스트됩니다.

 

🔍 관찰 포인트 — "왜 pandas로 집계하지 않나요?" pandas의 groupby로 집계하면 코드는 짧아집니다. 하지만 그러면 유스케이스가 pandas에 의존하게 되어, (1) 도메인/유스케이스의 순수성이 깨지고 (2) 테스트할 때 pandas를 끌어와야 합니다. 지금 규모(수천~수만 행)에서는 순수 파이썬 집계가 충분히 빠르고, 의존이 없어 테스트가 가볍습니다. 만약 수백만 행이 되면, "pandas로 집계하는 분석 어댑터"를 따로 만들어 끼우면 됩니다(11교시에서 논의). 핵심은 성능 최적화가 필요해도 유스케이스의 계약은 그대로 둔다는 것입니다.

📖 개념 2 — 무엇을 계산할 것인가

분석 유스케이스가 만들어 낼 결과는 AnalysisResult라는 도메인 객체이며, 채울 지표는 다음과 같습니다.

지표	의미	자료형
total_revenue	전체 매출 합계	float
total_quantity	전체 판매 수량	int
period_start/end	데이터의 시작·끝 날짜	date
by_product	상품별 매출 합계	dict[str, float]
by_region	지역별 매출 합계	dict[str, float]
by_month	월(YYYY-MM)별 매출 합계	dict[str, float]

---

💻 준비 — AnalysisResult 도메인 객체 확인 (먼저 하세요)

분석 유스케이스(analyze_sales.py)와 그 테스트는 domain/models.py의 AnalysisResult 를 import 합니다. 이 객체는 4교시에서 뼈대를 만들었지만, 5교시에서 SalesRecord에 집중하다 보면 models.py에 빠져 있을 수 있습니다.

⚠️ 이 단계를 건너뛰면 Red 테스트 실행 시 다음 오류가 납니다: ImportError: cannot import name 'AnalysisResult' from 'sales_agent.domain.models'

 

src/sales_agent/domain/models.py를 열어 AnalysisResult가 있는지 확인하고, 없으면 SalesRecord 아래에 추가하세요. (이미 있다면 이 단계는 그냥 넘어갑니다.)

# src/sales_agent/domain/models.py 상단 import 확인
#   - dataclass 옆에 field 가 함께 import 되어 있어야 한다(없으면 추가).
#   - date 가 import 되어 있어야 한다.
from dataclasses import dataclass, field   # ← field 가 빠져 있으면 추가
from datetime import date


# SalesRecord 정의 아래에 AnalysisResult 를 추가한다.
@dataclass(frozen=True)
class AnalysisResult:
    """분석 산출물을 담는 불변 객체. 이 교시의 유스케이스가 값을 채운다.

    필드:
        total_revenue: 전체 매출 합계
        total_quantity: 전체 판매 수량 합계
        period_start / period_end: 데이터의 가장 이른/늦은 날짜
        by_product / by_region / by_month: 각 기준별 매출 합계 사전(dict)
    """

    total_revenue: float
    total_quantity: int
    period_start: date
    period_end: date
    # field(default_factory=dict): 인스턴스마다 '새로운 빈 dict'를 기본값으로 준다.
    # (= {} 로 쓰면 모든 인스턴스가 같은 dict를 공유하는 고전적 버그가 생긴다.)
    by_product: dict[str, float] = field(default_factory=dict)  # 상품명 → 매출 합계
    by_region: dict[str, float] = field(default_factory=dict)   # 지역명 → 매출 합계
    by_month: dict[str, float] = field(default_factory=dict)    # "YYYY-MM" → 매출 합계

    @property
    def top_products(self) -> list[tuple[str, float]]:
        """상품별 매출을 '매출 내림차순'으로 정렬한 (상품명, 매출) 목록을 돌려준다."""
        return sorted(self.by_product.items(), key=lambda kv: kv[1], reverse=True)

 

확인이 끝나면 도메인 모듈이 정상적으로 import 되는지 점검합니다.

uv run python -c "from sales_agent.domain.models import SalesRecord, AnalysisResult; print('models OK')"

models OK가 출력되면 준비 완료입니다.

 

🔍 관찰 포인트: 분석 유스케이스가 만들 "결과의 모양(AnalysisResult)"을 먼저 도메인에 정의해 둡니다. 유스케이스는 이 결과 객체를 채울 뿐이고, AnalysisResult 자체는 외부 라이브러리 없는 순수 도메인이라는 점에 주목하세요. (revenue처럼 top_products도 저장 필드가 아니라 @property로 계산합니다.)

---

🔴 Red — 분석 결과를 테스트로 명세

tests/test_analyze_sales.py. 핵심은 Fake 리포지토리로 입력을 고정하는 것입니다.

from datetime import date

from sales_agent.domain.models import SalesRecord
from sales_agent.usecases.analyze_sales import AnalyzeSalesUseCase


class FakeSalesRepository:
    """SalesRepository 포트를 만족하는 테스트용 가짜 리포지토리.

    파일을 읽지 않고 미리 준비한 레코드를 그대로 돌려준다.
    덕분에 'CSV 적재'와 무관하게 '분석 계산'만 격리해서 검증할 수 있다.
    """

    def __init__(self, records):
        self._records = records   # 테스트가 정한 입력 레코드를 보관

    def load(self):
        return self._records      # 포트 약속(load)을 그대로 흉내


def sample_records():
    """검증에 쓸 '고정 입력'. 기대 결과를 손으로 암산할 수 있게 숫자를 단순화했다."""
    return [
        # SalesRecord(날짜, 상품, 지역, 수량, 단가)  → 주석은 그 행의 매출액
        SalesRecord(date(2024, 1, 5), "노트북", "서울", 5, 1_000_000),  # 5 × 1,000,000 = 5,000,000
        SalesRecord(date(2024, 1, 6), "마우스", "부산", 10, 20_000),    # 10 × 20,000   =   200,000
        SalesRecord(date(2024, 2, 11), "노트북", "서울", 2, 1_000_000), # 2 × 1,000,000 = 2,000,000
    ]


def make_result():
    """가짜 리포지토리를 끼운 유스케이스를 실행해 분석 결과를 돌려주는 헬퍼."""
    repo = FakeSalesRepository(sample_records())   # 입력 고정
    return AnalyzeSalesUseCase(repo).execute()     # 포트로 주입 → 실행


def test_total_revenue():
    """총매출은 모든 레코드 매출액의 합이다."""
    # 5,000,000 + 200,000 + 2,000,000 = 7,200,000
    assert make_result().total_revenue == 7_200_000


def test_total_quantity():
    """총수량은 모든 레코드 수량의 합이다."""
    # 5 + 10 + 2 = 17
    assert make_result().total_quantity == 17


def test_period_range():
    """기간은 가장 이른 날짜부터 가장 늦은 날짜까지다."""
    result = make_result()
    assert result.period_start == date(2024, 1, 5)   # 최소 날짜
    assert result.period_end == date(2024, 2, 11)    # 최대 날짜


def test_by_product_aggregation():
    """상품별 매출이 합산된다(같은 '노트북' 두 건이 하나로 합쳐진다)."""
    by_product = make_result().by_product
    assert by_product["노트북"] == 7_000_000   # 5,000,000 + 2,000,000
    assert by_product["마우스"] == 200_000


def test_by_region_aggregation():
    """지역별 매출이 합산된다."""
    by_region = make_result().by_region
    assert by_region["서울"] == 7_000_000   # 노트북 두 건 모두 서울
    assert by_region["부산"] == 200_000


def test_by_month_aggregation():
    """월(YYYY-MM)별 매출이 합산된다."""
    by_month = make_result().by_month
    assert by_month["2024-01"] == 5_200_000   # 1월: 5,000,000 + 200,000
    assert by_month["2024-02"] == 2_000_000   # 2월: 2,000,000


def test_top_products_is_sorted():
    """상위 상품은 매출 내림차순으로 정렬된다(가장 큰 것이 맨 앞)."""
    top = make_result().top_products
    assert top[0] == ("노트북", 7_000_000)   # 1위는 노트북


def test_empty_records_raises():
    """레코드가 하나도 없으면 분석할 대상이 없으므로 오류를 던진다."""
    import pytest
    with pytest.raises(ValueError):
        # 빈 목록을 가진 가짜를 주입 → execute()에서 ValueError가 나야 정상
        AnalyzeSalesUseCase(FakeSalesRepository([])).execute()

 

🔍 관찰 포인트 1 — 입력을 "손으로 검산 가능한" 숫자로 만듭니다. 단가를 1,000,000과 20,000처럼 단순하게 둔 이유는, 사람이 암산으로 기대값을 검증할 수 있게 하기 위함입니다. 노트북 = 5,000,000 + 2,000,000 = 7,000,000. 테스트의 기대값(7_000_000)이 맞는지 직접 확인 가능합니다. 테스트의 기대값 자체가 틀리면 테스트가 무의미하므로, 검산 가능한 입력은 중요합니다.

 

🔍 관찰 포인트 2 — Fake 덕분에 파일이 0개입니다. 이 테스트 파일에는 CSV 파일 경로도, tmp_path도 없습니다. FakeSalesRepository가 포트를 만족하므로, 유스케이스는 "진짜 CSV"인지 "가짜"인지 구분하지 못한 채 정상 동작합니다. 이것이 의존성 역전이 주는 테스트 편의입니다.

uv run pytest tests/test_analyze_sales.py -v   # 🔴 실패 확인

---

🟢 Green — 유스케이스 구현 (순수 파이썬)

이제 위 테스트를 통과시키는 AnalyzeSalesUseCase를 src/sales_agent/usecases/analyze_sales.py에 직접 작성합니다. 규칙은 ▲생성자에서 SalesRepository(포트)를 주입받고 구체 어댑터를 import 하지 않을 것 ▲pandas 없이 순수 파이썬으로 집계할 것 ▲레코드가 비면 ValueError를 던질 것입니다.

 

src/sales_agent/usecases/analyze_sales.py:

from __future__ import annotations

# defaultdict(float): 처음 보는 키를 자동으로 0.0으로 초기화해 주는 사전.
# 덕분에 "키가 있으면 더하고, 없으면 0부터" 같은 조건문 없이 += 만으로 누적할 수 있다.
from collections import defaultdict

from sales_agent.domain.models import AnalysisResult
from sales_agent.usecases.ports import SalesRepository  # 추상(포트)만 import


class AnalyzeSalesUseCase:
    """매출 레코드를 분석해 AnalysisResult(지표 묶음)를 만드는 유스케이스.

    데이터 출처(CSV/DB/API)도, pandas도, 화면도 모른다. 오직 포트와 도메인만 안다.
    """

    def __init__(self, repository: SalesRepository) -> None:
        # 추상(포트)에만 의존한다 → CsvSalesRepository, FakeSalesRepository 등
        # load()를 가진 무엇이든 주입할 수 있다(의존성 역전).
        self._repository = repository

    def execute(self) -> AnalysisResult:
        """레코드를 적재한 뒤 각종 지표를 집계해 AnalysisResult로 반환한다."""
        records = self._repository.load()   # 포트를 통해 레코드 확보(출처는 모른 채)
        if not records:                     # 빈 목록이면 분석할 대상이 없음
            raise ValueError("분석할 매출 레코드가 없습니다.")

        # 기준별 매출 누적용 사전 3개(상품/지역/월). 모두 0.0에서 시작.
        by_product: dict[str, float] = defaultdict(float)
        by_region: dict[str, float] = defaultdict(float)
        by_month: dict[str, float] = defaultdict(float)

        total_revenue = 0.0
        total_quantity = 0
        for r in records:                              # 레코드를 한 번 순회하며 동시에 집계
            total_revenue += r.revenue                 # 전체 매출 누적(도메인의 계산값 사용)
            total_quantity += r.quantity               # 전체 수량 누적
            by_product[r.product] += r.revenue         # 상품별 매출 누적
            by_region[r.region] += r.revenue           # 지역별 매출 누적
            by_month[r.date.strftime("%Y-%m")] += r.revenue  # "2024-01" 형태 월 키로 누적

        dates = [r.date for r in records]              # 기간 계산용 날짜 모음
        return AnalysisResult(
            total_revenue=total_revenue,
            total_quantity=total_quantity,
            period_start=min(dates),                   # 가장 이른 날짜
            period_end=max(dates),                     # 가장 늦은 날짜
            # defaultdict를 일반 dict로 변환해 반환(불필요한 자동 생성 동작 차단).
            by_product=dict(by_product),
            by_region=dict(by_region),
            by_month=dict(by_month),
        )

 

🔍 관찰 포인트 — 이 파일에 없는 것을 확인하세요. import pandas도, import csv도, 파일 경로 문자열도 없습니다. 오직 도메인 객체와 포트만 사용합니다. 그래서 이 유스케이스는 데이터 출처가 CSV든 DB든 API든 동일하게 동작합니다. defaultdict(float)는 "처음 보는 키는 0.0에서 시작"하게 해 누적 합산을 간결하게 만들어 줍니다.

uv run pytest tests/test_analyze_sales.py -v   # 🟢 통과 확인

---

🔵 Refactor — 그리고 실제 CSV로 연결 확인

코드를 다듬은 뒤(중복 정리·이름 정돈), 6교시의 실제 CSV 어댑터와 7교시 유스케이스가 함께 동작하는지 빠르게 확인합니다.

💡 PowerShell에서 여러 줄 파이썬 실행하기 — here-string 여러 줄 파이썬 코드를 PowerShell에서 실행할 때는, 아래처럼 @' ... '@(여기-문자열)로 코드를 묶어 파이썬의 표준 입력으로 흘려보내는 방식이 안전합니다. @'로 시작하고 '@로 끝나며, 작은따옴표 버전은 $ 등을 그대로 글자로 취급합니다(변환 없음). 마지막의 python -는 "표준 입력으로 들어온 코드를 실행"하라는 뜻입니다.

@'
from sales_agent.adapters.csv_repository import CsvSalesRepository
from sales_agent.usecases.analyze_sales import AnalyzeSalesUseCase

repo = CsvSalesRepository("sample_data/sales_sample.csv")
result = AnalyzeSalesUseCase(repo).execute()

print("총매출:", f"{result.total_revenue:,.0f}원")
print("총수량:", result.total_quantity)
print("상위 상품:", result.top_products[:3])
print("제외된 행 수:", len(repo.skipped))
for s in repo.skipped:
    print("  -", s.line, "행 제외:", s.reason)
'@ | uv run python -

 

기대 출력(요지):

총매출: 6,739,000원        # 잘못된 6번째 행(수량 0) 제외 후 합계
...
제외된 행 수: 1
  - 7 행 제외: quantity must be > 0, got 0

 

🔍 관찰 포인트 — 두 계층이 포트를 통해 자연스럽게 맞물립니다. 방금 만든 유스케이스(AnalyzeSalesUseCase)는 코드를 한 줄도 바꾸지 않았는데, 6교시의 진짜 어댑터(CsvSalesRepository)를 받아 동작했습니다. 테스트에서는 Fake를, 실제에서는 진짜 어댑터를 끼운 것뿐입니다. 이것이 "조립으로 시스템을 구성한다"는 클린 아키텍처의 작동 방식입니다.

uv run pytest -v   # 전체 회귀 확인

---

✏️ 미니 실습 — 새 지표 1개를 TDD로 추가

"평균 객단가(= 총매출 / 총수량)"를 추가해 보세요. 순서를 지키는 것이 핵심입니다.

1. docs/test-list.md에 항목 추가: [ ] 평균 객단가 = 총매출 / 총수량
2. test_analyze_sales.py에 테스트 먼저 작성(Red)
3. AnalysisResult에 @property로 계산값 추가(Green)
4. uv run pytest -v로 확인 → 다듬기(Refactor)

🔍 관찰 포인트: 새 기능도 "테스트 먼저"입니다. 이 리듬이 반복되면, 기능 추가가 곧 "안전한 작업"이 됩니다.

 

🔒 진행 게이트

• 제출물(기록): 평균 객단가 테스트 2개(정상 + 수량 0) 통과 + average_order_value 속성.
• 이 결과가 있어야 다음 교시로 넘어갑니다.
• 정답·해설(별도 파일): 직접 풀어 본 뒤 solutions/미니실습정답_07.md에서 확인하세요.

---

✅ 체크포인트

• [ ] domain/models.py에 AnalysisResult가 있고 models OK가 출력된다
• [ ] 8개 분석 테스트가 모두 통과한다
• [ ] 유스케이스가 pandas/파일을 import 하지 않는다
• [ ] Fake 리포지토리로 파일 없이 분석을 검증했다
• [ ] 실제 CSV로 어댑터+유스케이스 연결을 확인했다
• [ ] 새 지표 1개를 TDD로 추가해 봤다

---

🛠️ 트러블슈팅

증상	원인	해결
ImportError: cannot import name 'AnalysisResult'	domain/models.py에 AnalysisResult가 없음	위 준비 단계에서 AnalysisResult를 추가. import에 field가 있는지도 확인
ImportError: cannot import name 'field' 또는 NameError: field	from dataclasses import 에 field 누락	from dataclasses import dataclass, field로 수정
here-string 실행이 멈춤	'@가 줄 맨 앞이 아님	'@는 반드시 줄의 첫 글자여야 함(앞 공백 금지)
float 합계 미세 오차	부동소수점 특성	통화는 정수(원) 단위로 다루거나 비교 시 허용오차 사용
by_month 키 형식	strftime 포맷	"%Y-%m" → 2024-01 형태 확인
유스케이스에 pandas가 끼어듦	규칙 위반	어댑터로 옮기고 유스케이스는 순수 유지

---

🔑 핵심 정리

• 유스케이스는 포트만 알고 외부(pandas·파일·화면)를 모른다 — 그래서 재사용·테스트가 쉽다.
• Fake 리포지토리로 분석 계산을 격리 검증했고, 같은 유스케이스에 진짜 어댑터를 끼워 실제로도 동작시켰다.
• 지금 규모엔 순수 파이썬 집계가 충분하며, 필요 시 어댑터 교체로 확장한다.

---

🗓️ 1일차 마무리

여기까지가 1일차입니다. 환경 구축 → 문제 분해 → 클린 아키텍처 → 에이전트 설계 → 첫 TDD → CSV 적재 → 분석 로직을, 모두 직접 손으로 TDD 리듬에 따라 만들었습니다. 2일차에는 같은 방식으로 차트·LLM·리포트·웹까지 완성하고(8~11교시), 그다음 그 모든 작업을 Copilot 바이브 코딩으로 빠르게 재현·확장(12·13교시)합니다.

다음 교시(2일차 시작): 분석 결과를 차트 이미지로 그리는 어댑터를 만들고, "부수효과가 있는 코드"를 테스트하는 법을 배웁니다.