feat(lps): 최저가 이력(price_history) — 트리거 기반 시계열 그래프

같은 상품 반복 검색 시 최저가를 스냅샷으로 적재 → 네이버/쿠팡/최종 3개 선 그래프.
배치 아님(조회된 상품만, 실제 검색 시각에 기록) — 트래픽/리소스 절약.

- price_history 테이블: product_code·triggered_at(X축)·naver/coupang/final 최저가+상세·outcome
- crud/price_history: record() + list_by_product(시각 오름차순)
- handler: AI 매칭 후 소스별 min + 전체 min 스냅샷 기록(_price_snapshot).
  found/not_found 기록, 네거티브 캐시 히트·기술실패는 미기록
- API: GET /v1/lps/products/{product_code}/history → 그래프 데이터(시각 오름차순)
- tests: 스냅샷 계산/기록·조회/핸들러 기록규칙/API → 전체 54/54

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lps/common/database/model/models.py

@@ -68,6 +68,40 @@ class search_negative(MAIN_BASE):
     created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()"))
 
 
+class price_history(MAIN_BASE):
+    """상품별 최저가 스냅샷(트리거 기반). 네이버/쿠팡/최종 최저가를 검색 시점마다 적재해
+    시계열 그래프(X=triggered_at, Y=가격, 3개 선)로 본다. 배치 아님 — 조회된 상품만 기록."""
+
+    @staticmethod
+    def DBType():
+        return DBType.MAIN.value
+
+    __tablename__ = "price_history"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))
+    product_code = Column(String(100), nullable=False)                                            # 상품 식별(조회 키)
+    job_id = Column(UUID(as_uuid=True), nullable=True)                                            # 검색 잡 연결(추적)
+    triggered_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()"))  # X축(검색 실행 시각)
+    outcome = Column(String(20), nullable=False)                                                  # found / not_found
+    matched_count = Column(Integer, nullable=True)                                                # AI 매칭 건수
+
+    naver_lowest = Column(Integer, nullable=True)                                                 # 네이버 최저가(같은 상품)
+    naver_name = Column(String(300), nullable=True)
+    naver_url = Column(Text, nullable=True)
+    coupang_lowest = Column(Integer, nullable=True)                                               # 쿠팡 최저가(같은 상품)
+    coupang_name = Column(String(300), nullable=True)
+    coupang_url = Column(Text, nullable=True)
+    final_lowest = Column(Integer, nullable=True)                                                 # 전체 최저가(Y축 핵심)
+    final_source = Column(String(20), nullable=True)                                              # 최종 최저가 소스
+
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()"))
+
+    __table_args__ = (
+        # 특정 상품 시계열 조회 최적화
+        Index("ix_price_history_product", "product_code", "triggered_at"),
+    )
+
+
 class bot_detection(MAIN_BASE):
     """봇 감지 이력 — '이 IP로 몇 번째 요청에서, 어떤 방식으로 차단됐나'를 축적해 패턴 분석.
     (예: SELECT avg(ip_request_no) → IP당 평균 몇 요청 만에 감지되는지)"""

lps/crud/price_history.py

@@ -0,0 +1,63 @@
+"""최저가 스냅샷 CRUD — 트리거 시점마다 기록하고, 상품별 시계열로 조회(그래프)."""
+
+from sqlalchemy import text
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import DBType, DBWRType
+
+_FIELDS = (
+    "product_code", "job_id", "outcome", "matched_count",
+    "naver_lowest", "naver_name", "naver_url",
+    "coupang_lowest", "coupang_name", "coupang_url",
+    "final_lowest", "final_source",
+)
+
+
+class PriceHistory:
+    DB = DBType.MAIN.value
+
+    async def record(self, event: dict):
+        """스냅샷 1건 저장. triggered_at 은 now()(관측 시각). 로깅 실패가 검색을 막지 않도록 호출부에서 예외 처리."""
+        sql = text("""
+            INSERT INTO price_history
+                (product_code, job_id, outcome, matched_count,
+                 naver_lowest, naver_name, naver_url,
+                 coupang_lowest, coupang_name, coupang_url,
+                 final_lowest, final_source)
+            VALUES
+                (:product_code, :job_id, :outcome, :matched_count,
+                 :naver_lowest, :naver_name, :naver_url,
+                 :coupang_lowest, :coupang_name, :coupang_url,
+                 :final_lowest, :final_source)
+        """)
+        params = {k: event.get(k) for k in _FIELDS}
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value)
+        try:
+            await s.execute(sql, params)
+            await s.commit()
+        except Exception:
+            await s.rollback()
+            raise
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)
+
+    async def list_by_product(self, product_code: str, limit: int = 100) -> list[dict]:
+        """상품의 최근 스냅샷을 시각 오름차순(그래프 플롯용)으로 반환. 최근 limit 건."""
+        sql = text("""
+            SELECT * FROM (
+                SELECT triggered_at, outcome, matched_count,
+                       naver_lowest, naver_name, naver_url,
+                       coupang_lowest, coupang_name, coupang_url,
+                       final_lowest, final_source
+                FROM price_history
+                WHERE product_code = :pc
+                ORDER BY triggered_at DESC
+                LIMIT :lim
+            ) t ORDER BY triggered_at ASC
+        """)
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            rows = (await s.execute(sql, {"pc": product_code, "lim": limit})).mappings().all()
+            return [dict(r) for r in rows]
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)

lps/router/v1/lps/protocol.py

@@ -45,3 +45,22 @@ class Res_JobStatus(Res_WebPacketProtocol):
 
 class Res_QueueStats(Res_WebPacketProtocol):
     counts: dict[str, int] = Field(default_factory=dict, description="상태별 잡 개수")
+
+
+class PricePoint(BaseModel):
+    triggered_at: str = Field(description="관측 시각(X축)")
+    outcome: str
+    matched_count: Optional[int] = None
+    naver: Optional[int] = Field(None, description="네이버 최저가")
+    coupang: Optional[int] = Field(None, description="쿠팡 최저가")
+    final: Optional[int] = Field(None, description="전체 최저가(Y축)")
+    final_source: Optional[str] = None
+    naver_name: Optional[str] = None
+    naver_url: Optional[str] = None
+    coupang_name: Optional[str] = None
+    coupang_url: Optional[str] = None
+
+
+class Res_PriceHistory(Res_WebPacketProtocol):
+    product_code: Optional[str] = None
+    points: list[PricePoint] = Field(default_factory=list, description="시각 오름차순 스냅샷(그래프용)")

lps/router/v1/lps/search.py

@@ -2,9 +2,11 @@
 
 from fastapi import APIRouter, Depends
 
+from fastapi import Query
+
 from router.v1.validator.dependencies import RemoveNoneResponse
 from services.lps_service import LpsService
-from router.v1.lps.protocol import Res_JobStatus, Res_QueueStats, Res_Search, Req_Search
+from router.v1.lps.protocol import Res_JobStatus, Res_PriceHistory, Res_QueueStats, Res_Search, Req_Search
 
 router = APIRouter(prefix="/v1/lps", tags=["LPS"], responses={404: {"description": "Not found"}})
 
@@ -37,3 +39,13 @@ async def job_status(job_id: str, service: LpsService = Depends()):
 )
 async def queue_stats(service: LpsService = Depends()):
     return RemoveNoneResponse(await service.stats())
+
+
+@router.get(
+    path="/products/{product_code}/history",
+    response_model=Res_PriceHistory,
+    summary="최저가 이력(그래프)",
+    description="상품의 트리거별 최저가 스냅샷(네이버/쿠팡/최종)을 시각 오름차순으로 반환. 가격 시계열 그래프용.",
+)
+async def price_history(product_code: str, limit: int = Query(100, ge=1, le=1000), service: LpsService = Depends()):
+    return RemoveNoneResponse(await service.price_history(product_code, limit))

lps/services/lps_service.py

@@ -10,9 +10,12 @@ from fastapi import Depends
 
 from common.enums import ErrorType, JobStatus, JobType
 from crud.job_crud import JobQueue
+from crud.price_history import PriceHistory
 from router.v1.lps.protocol import (
     EnqueuedItem,
+    PricePoint,
     Res_JobStatus,
+    Res_PriceHistory,
     Res_QueueStats,
     Res_Search,
     SearchItem,
@@ -23,8 +26,9 @@ _PRIORITY = {"new": 1, "single": 2, "negowiz": 3, "batch": 4}
 
 
 class LpsService:
-    def __init__(self, queue: JobQueue = Depends(JobQueue)):
+    def __init__(self, queue: JobQueue = Depends(JobQueue), history: PriceHistory = Depends(PriceHistory)):
         self.queue = queue
+        self.history = history
 
     async def submit_search(self, items: list[SearchItem]) -> Res_Search:
         res = Res_Search()
@@ -65,3 +69,20 @@ class LpsService:
         res = Res_QueueStats()
         res.counts = await self.queue.counts()
         return res
+
+    async def price_history(self, product_code: str, limit: int = 100) -> Res_PriceHistory:
+        res = Res_PriceHistory(product_code=product_code)
+        rows = await self.history.list_by_product(product_code, limit)
+        res.points = [
+            PricePoint(
+                triggered_at=r["triggered_at"].isoformat(timespec="seconds"),
+                outcome=r["outcome"],
+                matched_count=r["matched_count"],
+                naver=r["naver_lowest"], coupang=r["coupang_lowest"], final=r["final_lowest"],
+                final_source=r["final_source"],
+                naver_name=r["naver_name"], naver_url=r["naver_url"],
+                coupang_name=r["coupang_name"], coupang_url=r["coupang_url"],
+            )
+            for r in rows
+        ]
+        return res

lps/tests/test_lps_api.py

@@ -51,6 +51,24 @@ async def test_job_status_not_found(client, clean_jobs):
     assert r2.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND"
 
 
+async def test_price_history_endpoint(client, db_engine):
+    from crud.price_history import PriceHistory
+    async with db_engine.begin() as conn:
+        await conn.execute(text("TRUNCATE price_history"))
+    ph = PriceHistory()
+    await ph.record({"product_code": "GRAPH1", "outcome": "found", "final_lowest": 2500,
+                     "final_source": "naver", "naver_lowest": 2500, "coupang_lowest": 2700, "matched_count": 2})
+
+    r = await client.get("/v1/lps/products/GRAPH1/history")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["product_code"] == "GRAPH1"
+    assert len(body["points"]) == 1
+    pt = body["points"][0]
+    assert pt["final"] == 2500 and pt["naver"] == 2500 and pt["coupang"] == 2700 and pt["final_source"] == "naver"
+    assert "triggered_at" in pt
+
+
 async def test_queue_stats(client, clean_jobs):
     await client.post("/v1/lps/search", json={"data": [
         {"product_code": "A", "product_name": "x"},

lps/tests/test_price_history.py

@@ -0,0 +1,93 @@
+"""최저가 이력 — 소스별 min 스냅샷 계산 + 기록/조회 CRUD + 핸들러 기록 규칙."""
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from common.enums import JobType
+from crud.price_history import PriceHistory
+from services.search.contract import NormalizedProduct
+from worker.handlers import _price_snapshot, build_search_handler
+
+
+def _np(source, price, name=None):
+    return NormalizedProduct(source=source, name=name or f"{source}-{price}", price=price, detail_url=f"http://{source}/{price}")
+
+
+# ── 스냅샷 계산 (순수) ─────────────────────────────────────────────
+def test_snapshot_source_lowest_and_final():
+    matched = [_np("naver", 3000), _np("naver", 2000), _np("coupang", 2500)]
+    s = _price_snapshot(matched)
+    assert s["naver_lowest"] == 2000 and s["coupang_lowest"] == 2500
+    assert s["final_lowest"] == 2000 and s["final_source"] == "naver"
+    assert s["matched_count"] == 3
+
+
+def test_snapshot_single_source_only():
+    s = _price_snapshot([_np("coupang", 1500)])
+    assert s["naver_lowest"] is None and s["coupang_lowest"] == 1500
+    assert s["final_lowest"] == 1500 and s["final_source"] == "coupang"
+
+
+def test_snapshot_empty_is_all_null():
+    s = _price_snapshot([])
+    assert s["final_lowest"] is None and s["naver_lowest"] is None and s["matched_count"] == 0
+
+
+# ── CRUD (실 DB) ───────────────────────────────────────────────────
+@pytest_asyncio.fixture
+async def ph(db_engine):
+    async with db_engine.begin() as conn:
+        await conn.execute(text("TRUNCATE price_history"))
+    return PriceHistory()
+
+
+async def test_record_and_list_time_ordered(ph):
+    await ph.record({"product_code": "P1", "outcome": "found", "final_lowest": 2000, "final_source": "naver",
+                     "naver_lowest": 2000, "coupang_lowest": 2500, "matched_count": 3})
+    await ph.record({"product_code": "P1", "outcome": "found", "final_lowest": 1900, "final_source": "coupang",
+                     "naver_lowest": 2100, "coupang_lowest": 1900, "matched_count": 2})
+    await ph.record({"product_code": "P2", "outcome": "found", "final_lowest": 999})  # 다른 상품
+
+    points = await ph.list_by_product("P1")
+    assert len(points) == 2                                   # P2 제외
+    assert [p["final_lowest"] for p in points] == [2000, 1900]  # 시각 오름차순
+    assert points[0]["triggered_at"] <= points[1]["triggered_at"]
+
+
+# ── 핸들러 기록 규칙 ───────────────────────────────────────────────
+class _Rec:
+    def __init__(self): self.events = []
+    async def record(self, e): self.events.append(e)
+
+
+class _FakeAdapter:
+    def __init__(self, source, products): self.source = source; self._p = products
+    async def search(self, q, limit=40): return self._p
+
+
+class _Neg:
+    def __init__(self, neg): self._neg = neg
+    async def is_negative(self, k): return self._neg
+    async def put(self, *a, **k): pass
+
+
+def _job(**p):
+    p.setdefault("product_name", "x"); p.setdefault("product_code", "PC1")
+    return {"job_type": JobType.SEARCH.value, "attempts": 1, "job_id": "j1", "payload": p}
+
+
+async def test_handler_records_found_snapshot():
+    rec = _Rec()
+    adapters = {"naver": _FakeAdapter("naver", [_np("naver", 2000)]), "coupang": _FakeAdapter("coupang", [_np("coupang", 1800)])}
+    await build_search_handler(adapters, history=rec)(_job())
+    assert len(rec.events) == 1
+    e = rec.events[0]
+    assert e["product_code"] == "PC1" and e["outcome"] == "found"
+    assert e["final_lowest"] == 1800 and e["final_source"] == "coupang"
+
+
+async def test_handler_skips_record_on_negative_cache_hit():
+    rec = _Rec()
+    adapters = {"naver": _FakeAdapter("naver", [_np("naver", 100)])}
+    await build_search_handler(adapters, neg_cache=_Neg(True), history=rec)(_job())
+    assert rec.events == []  # 캐시 히트 → 새 관측 없음 → 미기록

lps/worker/handlers.py

@@ -17,11 +17,28 @@ import asyncio
 
 from common.enums import JobType
 from common.logger import LOG
-from services.search.contract import SearchAdapter
+from services.search.contract import SearchAdapter, NormalizedProduct
 from services.search.util import parse_price
 from services.pipeline.core import apply_filters, rank_result
 
 
+def _price_snapshot(matched: list[NormalizedProduct]) -> dict:
+    """매칭 목록에서 소스별 최저가 + 전체 최저가 스냅샷을 만든다(price_history 기록용)."""
+    def lowest(src):
+        items = [p for p in matched if p.source == src]
+        return min(items, key=lambda p: p.price) if items else None
+
+    n, c = lowest("naver"), lowest("coupang")
+    finals = [x for x in (n, c) if x]
+    f = min(finals, key=lambda p: p.price) if finals else None
+    return {
+        "matched_count": len(matched),
+        "naver_lowest": n.price if n else None, "naver_name": n.name if n else None, "naver_url": n.detail_url if n else None,
+        "coupang_lowest": c.price if c else None, "coupang_name": c.name if c else None, "coupang_url": c.detail_url if c else None,
+        "final_lowest": f.price if f else None, "final_source": f.source if f else None,
+    }
+
+
 def build_search_handler(
     adapters: dict[str, SearchAdapter],
     sources: list[str] | None = None,
@@ -31,12 +48,23 @@ def build_search_handler(
     keyword_gen=None,
     max_rounds: int = 3,
     neg_cache=None,
+    history=None,
 ):
     """검색 핸들러 생성.
     judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) /
-    neg_cache: NegativeCache(TTL not_found 캐시). 모두 선택 — 없으면 해당 단계 생략."""
+    neg_cache: NegativeCache(TTL not_found 캐시) / history: PriceHistory(최저가 스냅샷).
+    모두 선택 — 없으면 해당 단계 생략."""
     use = list(sources) if sources else list(adapters.keys())
 
+    async def _record_history(product_code: str, job_id, outcome: str, matched: list):
+        if history is None:
+            return
+        event = {"product_code": product_code, "job_id": job_id, "outcome": outcome, **_price_snapshot(matched)}
+        try:
+            await history.record(event)
+        except Exception as ex:
+            LOG.e_no_callstack(f"[history] 스냅샷 기록 실패(무시): {ex}")
+
     async def _search_round(query: str):
         """한 라운드: 모든 소스 동시 검색 → (products, per_source, tech_failed)."""
         results = await asyncio.gather(*[adapters[s].search(query, limit=limit) for s in use], return_exceptions=True)
@@ -99,6 +127,7 @@ def build_search_handler(
             if candidates:  # 찾음 → 조기 종료
                 result = rank_result(candidates, len(products), stages, top_n)
                 result.update(outcome="found", query=query, round=label, rounds_tried=rounds_done, sources=per_source)
+                await _record_history(cache_key, job.get("job_id"), "found", candidates)
                 return result
 
             if tech_failed:  # 0매칭인데 소스가 죽어 있었음 → '없음'이라 단정 불가 → 기술 재시도
@@ -109,6 +138,7 @@ def build_search_handler(
             await neg_cache.put(cache_key, reason=f"not_found after {rounds_done} rounds")
         result = rank_result([], 0, last_stages, top_n)
         result.update(outcome="not_found", query=base_query, rounds_tried=rounds_done, sources=last_sources)
+        await _record_history(cache_key, job.get("job_id"), "not_found", [])
         return result
 
     return handler

lps/worker_main.py

@@ -13,6 +13,7 @@ from config.server_configs import web_server_config, openai_config
 from crud.job_crud import JobQueue
 from crud.negative_cache import NegativeCache
 from crud.bot_detection import BotDetectionLog
+from crud.price_history import PriceHistory
 from services.search.proxy import DecodoProxy
 from services.search.coupang.adapter import CoupangAdapter
 from services.search.naver.adapter import NaverAdapter
@@ -42,7 +43,10 @@ async def main(concurrency: int = 1):
     judge = SimilarityJudge() if has_openai else None
     keyword_gen = KeywordGenerator() if has_openai else None
     LOG.i(f"AI(판정+검색어생성): {'ON' if has_openai else 'OFF(키 없음)'}")
-    handler = build_search_handler(adapters, judge=judge, keyword_gen=keyword_gen, neg_cache=NegativeCache())
+    handler = build_search_handler(
+        adapters, judge=judge, keyword_gen=keyword_gen,
+        neg_cache=NegativeCache(), history=PriceHistory(),
+    )
 
     stop = asyncio.Event()
     listeners: list[JobListener] = []