feat(negodata): LPS 최저가 동기화 배치 — 요청(API)·수집(lps_db)·반영(items)

설계(2026-07-10 결정): 요청은 LPS API(dedupe·워커알림 보존), 결과는
lps_db.price_history 직접 읽기(읽기전용 엔진). product_code=item_id
로 결과가 자동 매핑된다.

- 잡③ request_lps_searches(매일 04:00 KST): 24h 이상 미수집 상품을
  최대 500건 enqueue(job_type=batch, 100건/콜 청크) — 비용 발생 잡
- 잡④ sync_lps_results(5분): 워터마크(max crawl_end_time) 증분 수집
  → item_internet_lowest_prices append(성공/실패 모두) + 성공분 최신값
  items.internet_lowest_price 박제(+yn). 한 트랜잭션(부분반영 방지)
- 스캔 바닥 3일 — 비uuid·미존재 상품 행이 워터마크를 못 올려도
  재스캔 범위 유한
- ⚠️ 시각은 aware UTC 통일 — naive 를 timestamptz 파라미터로 넘기면
  PG 세션 타임존(KST) 해석으로 9시간 어긋남(중복 수집 실측 버그 수정)
- LowestPriceWebsite 코드 enum(naver=1 coupang=2 …), iilp ORM 모델,
  lps_base_url config(+LPS_BASE_URL env)

검증: 실상품 1건 e2e(요청→크롤 found 15,000원→이력+박제 반영),
멱등성(재실행 빈 카운터), 비uuid 12건 스킵, 대상선정 쿼리 500건 상한

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

negodata/backend/common/database/model/models.py

@@ -111,6 +111,26 @@ class items(MainTableMixin, MAIN_BASE):
     delivery_fee_yn = Column(Boolean, nullable=True)
 
 
+class item_internet_lowest_prices(MainTableMixin, MAIN_BASE):
+    """인터넷 최저가 수집 이력(시도 단위, 성공/실패 모두 기록).
+    확정 대표값은 items.internet_lowest_price 에 박제된다(N건 이력 → 1건 대표 패턴).
+    LPS 동기화 배치(services/lps_sync_service)가 lps_db.price_history 를 읽어 채운다."""
+
+    __tablename__ = "item_internet_lowest_prices"
+    __table_args__ = {"schema": "partner"}
+
+    lp_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    item_id = Column(UUID(as_uuid=True), nullable=False, index=True)
+
+    lp_price = Column(BigInteger, nullable=True)         # 수집한 최저가(원). 실패면 NULL
+    website = Column(SmallInteger, nullable=False)       # LowestPriceWebsite 코드
+    success_yn = Column(Boolean, nullable=False)
+    fail_reason = Column(String(100), nullable=True)     # 실패 사유(예: not_found)
+    ai_model = Column(SmallInteger, nullable=True)       # (예약) AI 모델 코드 — LPS 계약엔 미포함
+    crawl_duration_ms = Column(Integer, nullable=True)   # (예약) 수집 소요 — LPS 계약엔 미포함
+    crawl_end_time = Column(DateTime(timezone=True), nullable=False)  # 수집 완료 시각(=price_history.created_at, 워터마크 기준)
+
+
 class suppliers(MainTableMixin, MAIN_BASE):
     __tablename__ = "suppliers"
     __table_args__ = {"schema": "partner"}

negodata/backend/common/enums.py

@@ -265,6 +265,25 @@ class SupplierType(CodeEnum):
     SOLE_AGENCY = 3   # 총판
 
 
+class LowestPriceWebsite(CodeEnum):
+    """partner.item_internet_lowest_prices.website — 최저가 수집 사이트 코드.
+    LPS(lps_db.price_history.final_source)의 소스 문자열을 코드값으로 매핑한다."""
+
+    NAVER = 1
+    COUPANG = 2
+    GMARKET = 3
+    AUCTION = 4
+    ST11 = 5
+    ETC = 99
+
+    @classmethod
+    def from_source(cls, source) -> "LowestPriceWebsite":
+        return {
+            "naver": cls.NAVER, "coupang": cls.COUPANG,
+            "gmarket": cls.GMARKET, "auction": cls.AUCTION, "st11": cls.ST11,
+        }.get((source or "").lower(), cls.ETC)
+
+
 class CardUsageType(CodeEnum):
     """nego_cards/wild_cards.usage_type 코드값. 협상카드 사용 범위(신규/재 견적·협상 양쪽 적용).
     공통=모두 적용(기본), 신규견적전용, 재견적전용."""

negodata/backend/config/config_models.py

@@ -10,6 +10,7 @@ class WebServerConfig(ConfigModel):
     client_url: str = ""
     nego_chat_url: str = "http://localhost:3300"
     agent_base_url: str = "http://localhost:9500"   # 협상 agent(9500). 공용 카탈로그 변경 알림용.
+    lps_base_url: str = "http://localhost:9600"     # 인터넷 최저가 검색 LPS(9600). 검색요청 enqueue 용.
 
 
 class LogConfig(ConfigModel):

negodata/backend/config/server_configs.py

@@ -59,3 +59,7 @@ def _apply_lps_db_env_override(cfg: LpsDBConfig):
 
 
 _apply_lps_db_env_override(lps_db_config)
+
+# LPS API 주소 env override (도커: http://lps-api:9600 또는 host.docker.internal:9600)
+if os.environ.get("LPS_BASE_URL"):
+    web_server_config.lps_base_url = os.environ["LPS_BASE_URL"]

negodata/backend/crud/lps_sync_crud.py

@@ -0,0 +1,145 @@
+"""LPS 동기화 CRUD — negosium(main) 쪽 조회/반영 + lps_db(읽기전용) 결과 조회.
+
+경계 규칙
+- lps_db 는 LPS 서비스 소유라 ORM 모델을 만들지 않는다 — anchoring reader 선례대로
+  raw table() 로 **price_history 만** 읽는다(그 외 테이블은 LPS 내부 구현으로 간주).
+- 워터마크는 별도 상태 테이블 없이 partner.item_internet_lowest_prices 의
+  max(crawl_end_time) 을 쓴다(crawl_end_time = price_history.created_at 을 그대로 보존).
+  반영이 중간에 실패해도 다음 tick 이 같은 지점부터 다시 읽는다(자기 치유).
+"""
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+from sqlalchemy import column, func, or_, select, table, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import item_internet_lowest_prices, items
+from common.enums import ErrorType
+from common.logger import LOG
+
+# lps_db.price_history — LPS 가 검색 1건마다 남기는 append-only 스냅샷(읽기 계약).
+_price_history = table(
+    "price_history",
+    column("product_code"),   # = negosium items.item_id (요청 시 그대로 넣는다)
+    column("outcome"),        # found | not_found
+    column("final_lowest"),   # 전체 최저가(원)
+    column("final_source"),   # naver | coupang | (폴백몰)
+    column("created_at"),
+)
+
+
+class ILpsSyncCRUD(ABC):
+    @abstractmethod
+    async def watermark(self, cdb: AsyncSession) -> Tuple[ErrorType, object]:
+        pass
+
+    @abstractmethod
+    async def stale_items(self, cdb: AsyncSession, cutoff, limit) -> Tuple[ErrorType, list]:
+        pass
+
+    @abstractmethod
+    async def existing_item_ids(self, cdb: AsyncSession, item_ids: list) -> Tuple[ErrorType, set]:
+        pass
+
+    @abstractmethod
+    async def fetch_results_since(self, ldb: AsyncSession, since) -> Tuple[ErrorType, list]:
+        pass
+
+    @abstractmethod
+    async def add_history_rows(self, cdb: AsyncSession, rows: list) -> ErrorType:
+        pass
+
+    @abstractmethod
+    async def update_item_lowest(self, cdb: AsyncSession, item_id, price) -> ErrorType:
+        pass
+
+
+class LpsSyncCRUD(ILpsSyncCRUD):
+    # ---- negosium(main) 조회 -------------------------------------------
+    async def watermark(self, cdb: AsyncSession):
+        """수집 이력의 최신 crawl_end_time — 증분 동기화 기준점. 이력이 없으면 None(전체 수집)."""
+        try:
+            q = select(func.max(item_internet_lowest_prices.crawl_end_time)).where(
+                item_internet_lowest_prices.deleted == False  # noqa: E712
+            )
+            return ErrorType.SUCCESS, (await cdb.execute(q)).scalar()
+        except Exception as ex:
+            LOG.e_no_callstack(f"[lps-sync] watermark 조회 실패: {ex}")
+            return ErrorType.DB_RUN_FAILED, None
+
+    async def stale_items(self, cdb: AsyncSession, cutoff, limit: int):
+        """재검색 대상 상품 — 수집 이력이 없거나 마지막 수집이 cutoff 이전인 활성 상품.
+        반환 행: (item_id, name, model_name, spec, manufacturer, price)"""
+        try:
+            latest = (
+                select(
+                    item_internet_lowest_prices.item_id.label("item_id"),
+                    func.max(item_internet_lowest_prices.crawl_end_time).label("last_ts"),
+                )
+                .where(item_internet_lowest_prices.deleted == False)  # noqa: E712
+                .group_by(item_internet_lowest_prices.item_id)
+                .subquery()
+            )
+            q = (
+                select(items.item_id, items.name, items.model_name, items.spec, items.manufacturer, items.price)
+                .join(latest, items.item_id == latest.c.item_id, isouter=True)
+                .where(
+                    items.deleted == False,  # noqa: E712
+                    or_(latest.c.last_ts.is_(None), latest.c.last_ts < cutoff),
+                )
+                .order_by(latest.c.last_ts.asc().nullsfirst())  # 오래된 것부터(이력 없는 신규 최우선)
+                .limit(limit)
+            )
+            return ErrorType.SUCCESS, (await cdb.execute(q)).all()
+        except Exception as ex:
+            LOG.e_no_callstack(f"[lps-sync] stale_items 조회 실패: {ex}")
+            return ErrorType.DB_RUN_FAILED, []
+
+    async def existing_item_ids(self, cdb: AsyncSession, item_ids: list):
+        """전달된 id 중 실재하는 활성 상품 id 집합 — 결과 반영 전 매핑 검증용."""
+        if not item_ids:
+            return ErrorType.SUCCESS, set()
+        try:
+            q = select(items.item_id).where(
+                items.item_id.in_(item_ids),
+                items.deleted == False,  # noqa: E712
+            )
+            return ErrorType.SUCCESS, {r[0] for r in (await cdb.execute(q)).all()}
+        except Exception as ex:
+            LOG.e_no_callstack(f"[lps-sync] existing_item_ids 조회 실패: {ex}")
+            return ErrorType.DB_RUN_FAILED, set()
+
+    # ---- lps_db(읽기전용) 조회 -----------------------------------------
+    async def fetch_results_since(self, ldb: AsyncSession, since):
+        """워터마크 이후의 price_history 증분. since 가 None 이면 전체(첫 동기화)."""
+        try:
+            q = select(
+                _price_history.c.product_code,
+                _price_history.c.outcome,
+                _price_history.c.final_lowest,
+                _price_history.c.final_source,
+                _price_history.c.created_at,
+            ).order_by(_price_history.c.created_at.asc())
+            if since is not None:
+                q = q.where(_price_history.c.created_at > since)
+            return ErrorType.SUCCESS, (await ldb.execute(q)).all()
+        except Exception as ex:
+            LOG.e_no_callstack(f"[lps-sync] price_history 조회 실패: {ex}")
+            return ErrorType.DB_RUN_FAILED, []
+
+    # ---- negosium(main) 반영 (execute_lambda_run 안에서 호출) -----------
+    async def add_history_rows(self, cdb: AsyncSession, rows: list) -> ErrorType:
+        """수집 이력 일괄 insert. rows = item_internet_lowest_prices ORM 객체 리스트."""
+        if not rows:
+            return ErrorType.SUCCESS
+        return await DB_SESSION_MNG.insert(cdb, rows, err_msg="lps-sync history insert 실패")
+
+    async def update_item_lowest(self, cdb: AsyncSession, item_id, price) -> ErrorType:
+        """상품 대표 최저가 박제 — items.internet_lowest_price + yn 플래그."""
+        q = (
+            update(items)
+            .where(items.item_id == item_id, items.deleted == False)  # noqa: E712
+            .values(internet_lowest_price=price, internet_lowest_price_yn=True)
+        )
+        return await DB_SESSION_MNG.add(cdb, q, err_msg=f"lps-sync 최저가 반영 실패 item={item_id}")

negodata/backend/scheduler/__init__.py

@@ -4,6 +4,8 @@
 
 잡 ① close_expired_quotations  : 5분마다(KST) — 마감시각 지난 견적 마감
 잡 ② close_negotiated_quotations: 5분마다(KST) — 모든 세션 협상 끝난 견적 즉시 마감(타입 무관)
+잡 ③ request_lps_searches      : 매일 04:00(KST) — 갱신 오래된 상품을 LPS 에 검색 요청(비용 발생 잡)
+잡 ④ sync_lps_results          : 5분마다(KST) — lps_db 결과 증분 수집 → 이력 append + 상품 최저가 박제
 """
 import os
 
@@ -51,8 +53,26 @@ def start_scheduler():
         misfire_grace_time=600,
         max_instances=1,
     )
+    # 잡 ③ LPS 검색 요청(비용 발생) — 새벽 1회. LPS 미설정 환경이면 잡 내부에서 스킵.
+    _scheduler.add_job(
+        jobs.request_lps_searches,
+        CronTrigger(hour=4, minute=0),
+        id="request_lps_searches",
+        coalesce=True,
+        misfire_grace_time=3600,  # 재기동 등으로 놓쳐도 1시간 내면 실행
+        max_instances=1,
+    )
+    # 잡 ④ LPS 결과 수집·반영 — 증분·멱등이라 잦아도 안전
+    _scheduler.add_job(
+        jobs.sync_lps_results,
+        CronTrigger(minute="*/5"),
+        id="sync_lps_results",
+        coalesce=True,
+        misfire_grace_time=600,
+        max_instances=1,
+    )
     _scheduler.start()
-    LOG.i("[scheduler] started (KST, both every 5min)")
+    LOG.i("[scheduler] started (KST: 견적마감 2잡 5분 · LPS 요청 04:00 · LPS 수집 5분)")
 
 
 def shutdown_scheduler():

negodata/backend/scheduler/jobs.py

@@ -79,3 +79,41 @@ async def close_negotiated_quotations() -> int:
     if results:
         LOG.i(f"[scheduler] close_negotiated: {_format_results(results)}")
     return sum(results.values())
+
+
+# ---- LPS(인터넷 최저가) 동기화 ------------------------------------------
+
+async def request_lps_searches() -> int:
+    """[잡③] 갱신이 오래된 상품을 LPS 에 검색 요청(enqueue). 매일 새벽 1회.
+    비용이 발생하는 잡(상품당 ~$0.004) — 주기·상한은 lps_sync_service 상수로 관리.
+    LPS 미설정 환경이면 조용히 스킵(available=False)."""
+    from services.lps_sync_service import LpsSyncService
+
+    service = LpsSyncService()
+    if not service.available():
+        return 0
+    results = await service.request_stale_searches()
+    if results:
+        LOG.i(
+            f"[scheduler] lps_request: 접수 {results['accepted']} / 활성중복 {results['duplicated']} / "
+            f"이름없음 {results['skipped_no_name']} / HTTP오류 {results['http_error']}"
+        )
+    return results["accepted"]
+
+
+async def sync_lps_results() -> int:
+    """[잡④] lps_db.price_history 증분을 읽어 수집 이력 append + 상품 대표 최저가 박제. 5분마다.
+    워터마크(=이력의 max crawl_end_time) 기준 증분이라 멱등 — 실패 tick 은 다음 tick 이 흡수."""
+    from services.lps_sync_service import LpsSyncService
+
+    service = LpsSyncService()
+    if not service.available():
+        return 0
+    results = await service.sync_results()
+    if results:
+        LOG.i(
+            f"[scheduler] lps_sync: found {results['found']} / not_found {results['not_found']} / "
+            f"상품반영 {results['items_updated']} / 비uuid스킵 {results['skipped_not_uuid']} / "
+            f"미존재상품 {results['skipped_unknown_item']} / 트랜잭션오류 {results['tx_error']}"
+        )
+    return results["items_updated"]

negodata/backend/services/lps_sync_service.py

@@ -0,0 +1,168 @@
+"""LPS(인터넷 최저가 검색) 동기화 서비스 — 요청·수집·반영의 도메인 로직.
+
+흐름 (요청은 API, 결과는 DB — 2026-07-10 설계 결정)
+  ① 요청: 갱신이 오래된 상품을 골라 LPS API(POST /v1/lps/search)로 enqueue.
+     product_code = items.item_id(uuid 문자열) — 이걸로 결과가 자동 매핑된다.
+     enqueue 를 DB insert 로 하지 않는 이유: LPS 의 활성중복 dedupe·pg_notify 워커 깨움을 우회하게 됨.
+  ② 수집: lps_db.price_history(읽기전용)를 워터마크(max crawl_end_time) 증분으로 읽는다.
+  ③ 반영: 이력은 partner.item_internet_lowest_prices 에 append(성공/실패 모두),
+     성공분의 상품별 최신값을 items.internet_lowest_price 에 박제(+yn).
+
+방어
+- LPS 미설정 환경(lps_db 미등록)이면 조용히 스킵 — 연동은 선택 기능, 본 서비스를 못 멈춘다.
+- product_code 가 uuid 가 아니거나(item_id 아님 — 예: LPS 자체 부하테스트 잡) 상품이 없으면 스킵.
+- 반영은 한 트랜잭션(execute_lambda_run) — 부분 반영으로 워터마크가 오염되지 않는다.
+"""
+import uuid
+from collections import Counter
+from datetime import timedelta, timezone
+
+import httpx
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import item_internet_lowest_prices
+from common.enums import DBType, DBWRType, ErrorType, LowestPriceWebsite
+from common.logger import LOG
+from config.server_configs import web_server_config
+from crud.lps_sync_crud import ILpsSyncCRUD, LpsSyncCRUD
+
+# 재검색 주기·요청 배치 크기 기본값. 상품당 실측 비용 ~$0.004 이므로
+# (상품 수 × 24h 주기)로 월 비용이 바로 계산된다. 필요 시 여기만 조정.
+REFRESH_HOURS = 24          # 마지막 수집이 이보다 오래된 상품만 재요청
+REQUEST_BATCH_LIMIT = 500   # 요청 잡 1회가 enqueue 하는 최대 상품 수(비용 상한)
+ENQUEUE_CHUNK = 100         # LPS API 1콜에 담는 상품 수
+SCAN_FLOOR_DAYS = 3         # price_history 스캔 하한(일) — 스킵행(비uuid 등)은 워터마크를
+                            # 못 올리므로, 바닥 없이는 같은 행을 영원히 재스캔하게 된다
+
+
+def _utc_now_aware():
+    """aware UTC now — 이 서비스는 timestamptz 비교/워터마크에 쓰므로 GTime(naive)이 아닌 aware 를 쓴다."""
+    from datetime import datetime
+    return datetime.now(timezone.utc)
+
+
+class LpsSyncService:
+    def __init__(self, crud: ILpsSyncCRUD = None):
+        self.crud = crud or LpsSyncCRUD()
+
+    @staticmethod
+    def available() -> bool:
+        """LPS 연동 활성 여부 — lps_db 가 등록된 환경에서만 배치가 돈다."""
+        return DB_SESSION_MNG.is_registered(DBType.LPS.value)
+
+    # ---- ① 요청: 오래된 상품을 LPS 에 검색 enqueue ----------------------
+    async def request_stale_searches(self) -> Counter:
+        results = Counter()
+        if not self.available():
+            return results
+
+        cutoff = _utc_now_aware() - timedelta(hours=REFRESH_HOURS)   # timestamptz 비교 — aware 필수(위 sync_results 주석)
+
+        err, rows = await DB_SESSION_MNG.execute_lambda(
+            DBType.MAIN.value, DBWRType.DB_READ.value,
+            lambda s: self.crud.stale_items(s, cutoff, REQUEST_BATCH_LIMIT),
+        )
+        if err != ErrorType.SUCCESS or not rows:
+            return results
+
+        payload_items = []
+        for item_id, name, model_name, spec, manufacturer, price in rows:
+            if not (name or "").strip():
+                results["skipped_no_name"] += 1
+                continue
+            payload_items.append({
+                "product_code": str(item_id),
+                "product_name": name,
+                "job_type": "batch",
+                "model": model_name or "",
+                "specification": spec or "",
+                "company": manufacturer or "",
+                "price": str(price) if price else "",
+            })
+
+        base = web_server_config.lps_base_url.rstrip("/")
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            for i in range(0, len(payload_items), ENQUEUE_CHUNK):
+                chunk = payload_items[i:i + ENQUEUE_CHUNK]
+                try:
+                    r = await client.post(f"{base}/v1/lps/search", json={"data": chunk})
+                    r.raise_for_status()
+                    body = r.json()
+                    results["accepted"] += int(body.get("accepted", 0))
+                    results["duplicated"] += sum(1 for it in body.get("items", []) if it.get("duplicated"))
+                except Exception as ex:
+                    # LPS 다운은 이 배치의 실패일 뿐 — 다음 tick 이 같은 대상(여전히 stale)을 다시 요청한다.
+                    results["http_error"] += 1
+                    LOG.w(f"[lps-sync] 검색요청 실패(청크 {i // ENQUEUE_CHUNK}): {type(ex).__name__}: {ex}")
+        return results
+
+    # ---- ②+③ 수집·반영: price_history 증분 → 이력 append + 대표값 박제 --
+    async def sync_results(self) -> Counter:
+        results = Counter()
+        if not self.available():
+            return results
+
+        err, wm = await DB_SESSION_MNG.execute_lambda(
+            DBType.MAIN.value, DBWRType.DB_READ.value, lambda s: self.crud.watermark(s))
+        if err != ErrorType.SUCCESS:
+            return results
+
+        # 스캔 시작점 = max(워터마크, 최근 N일 바닥) — 반영 안 되는 행(비uuid·미존재 상품)이
+        # 섞여 있어도 재스캔 범위가 유한하게 유지된다(N일 지나면 자연 소멸).
+        # ⚠️ 이 경로의 시각은 전부 aware UTC 로 다룬다. naive 를 timestamptz 파라미터로 넘기면
+        # PG 가 세션 타임존(KST)으로 해석해 9시간 어긋난 증분을 읽는다(중복 수집 실측 버그).
+        floor = _utc_now_aware() - timedelta(days=SCAN_FLOOR_DAYS)
+        since = max(wm, floor) if wm is not None else floor
+
+        err, rows = await DB_SESSION_MNG.execute_lambda(
+            DBType.LPS.value, DBWRType.DB_READ.value, lambda s: self.crud.fetch_results_since(s, since))
+        if err != ErrorType.SUCCESS or not rows:
+            return results
+
+        # product_code(uuid=item_id) 검증 — LPS 부하테스트 등 비상품 코드는 조용히 스킵
+        parsed = []
+        for code, outcome, final_lowest, final_source, created_at in rows:
+            try:
+                parsed.append((uuid.UUID(code), outcome, final_lowest, final_source, created_at))
+            except (ValueError, AttributeError, TypeError):
+                results["skipped_not_uuid"] += 1
+
+        err, existing = await DB_SESSION_MNG.execute_lambda(
+            DBType.MAIN.value, DBWRType.DB_READ.value,
+            lambda s: self.crud.existing_item_ids(s, list({p[0] for p in parsed})),
+        )
+        if err != ErrorType.SUCCESS:
+            return results
+
+        history_rows, latest_found = [], {}   # latest_found: item_id → (created_at, price)
+        for item_id, outcome, final_lowest, final_source, created_at in parsed:
+            if item_id not in existing:
+                results["skipped_unknown_item"] += 1
+                continue
+            found = outcome == "found" and final_lowest is not None
+            history_rows.append(item_internet_lowest_prices(
+                item_id=item_id,
+                lp_price=final_lowest if found else None,
+                website=LowestPriceWebsite.from_source(final_source).value,
+                success_yn=found,
+                fail_reason=None if found else (outcome or "unknown")[:100],
+                crawl_end_time=created_at,   # 워터마크 기준값 — price_history.created_at 그대로 보존
+            ))
+            results["found" if found else "not_found"] += 1
+            if found and (item_id not in latest_found or created_at > latest_found[item_id][0]):
+                latest_found[item_id] = (created_at, final_lowest)
+
+        if not history_rows:
+            return results
+
+        funcs = [lambda s, r=history_rows: self.crud.add_history_rows(s, r)]
+        funcs += [
+            (lambda s, i=item_id, p=price: self.crud.update_item_lowest(s, i, p))
+            for item_id, (_, price) in latest_found.items()
+        ]
+        err = await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value] * len(funcs), funcs)
+        if err != ErrorType.SUCCESS:
+            LOG.e_no_callstack(f"[lps-sync] 반영 트랜잭션 실패: {err.name} (다음 tick 재시도)")
+            return Counter({"tx_error": 1})
+        results["items_updated"] = len(latest_found)
+        return results