feat(lps): IP 선제 로테이션 — 요청 예산·포트 쿨다운·ip_session 관측

쿠팡 크롤 IP 를 '막힐 때까지' 쓰던 방식을 '막히기 전에 교체'로 전환한다.

- 요청 예산(LPS_IP_REQUEST_BUDGET, 기본 3): IP당 요청 수가 예산에 닿으면
  차단 전에 선제 회전. 실측상 5회 부근 차단 이력이 있어 보수적으로 3회.
  선제 교체된 포트는 평판이 깨끗해 로테이션 복귀 시 재사용된다.
- 포트 쿨다운(LPS_PORT_COOLDOWN_SEC, 기본 max(sticky,30분)): 차단 감지·
  전송오류 포트는 격리하고 _port() 가 건너뛴다. 전 포트 쿨다운이면 만료
  임박 포트 사용(가용성 우선). 포트 수는 config 범위에서 동적 산출.
- 차단 재시도 소진 시에도 회전 예약 — 불탄 포트로 다음 검색을 하지 않음.
- ip_session 테이블 신설: 세션마다 요청 수·성공/차단·종료 사유(budget/
  block/proxy_error/window/idle/shutdown)를 기록. bot_detection 과 달리
  무사 종료도 남아 예산 상한 튜닝의 원천 데이터가 된다(쿼리 database.md).
  models.py·migrations·init.sql(lps_db 섹션) 동행 갱신, dev DB 적용 완료.
- 테스트 17건 추가(쿨다운·예산 판정·세션 기록·CRUD), 전체 126 passed.

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

docker-compose.yml

@@ -164,6 +164,8 @@ services:
       LPS_PROFILE_DIR: /profiles       # Chrome 프로필을 영속 볼륨에 → 재시작해도 cf_clearance 유지(재웜업 회피)
       # LPS_FALLBACKS: "gmarket,auction,st11"  # 오픈마켓 폴백(기본 OFF — 켜기 전 라이브 스모크로 셀렉터 점검)
       # LPS_JOB_DEADLINE_SEC: "300"    # 잡 1건 처리 상한(행 방어) — 기본 300s
+      # LPS_IP_REQUEST_BUDGET: "3"     # IP당 요청 예산 — 도달 시 차단 전 선제 회전(0=비활성). 기본 3
+      # LPS_PORT_COOLDOWN_SEC: "1800"  # 차단 감지된 프록시 포트 격리 시간 — 기본 max(sticky, 30분)
       # ── 시크릿 주입(이미지엔 없음 — 필수). 리포 루트 .env 에 값 채움(.env.example 참고) ──
       OPENAI_API_KEY: ${OPENAI_API_KEY:-}          # 비면 AI 판정 OFF
       NAVER_KEYS: ${NAVER_KEYS:-}                  # "id1:secret1,id2:secret2" — 비면 네이버 검색 실패

lps/common/database/model/models.py

@@ -105,6 +105,33 @@ class price_history(MAIN_BASE):
     )
 
 
+class ip_session(MAIN_BASE):
+    """IP(프록시 포트) 세션 종료 이력 — '이 IP 로 몇 번 요청하고 어떻게 끝났나'를 매 세션 기록.
+    bot_detection 은 차단된 세션만 남지만 여기엔 무사 종료도 남아, 요청 예산(LPS_IP_REQUEST_BUDGET)
+    상한 튜닝의 원천 데이터가 된다. (예: end_reason='block' 의 requests 분포 → 안전 상한 산출)"""
+
+    @staticmethod
+    def DBType():
+        return DBType.MAIN.value
+
+    __tablename__ = "ip_session"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))
+    source = Column(String(20), nullable=False)                                                   # coupang 등
+    proxy_port = Column(Integer, nullable=True)                                                   # 사용 포트(=IP 세션), 프록시 미사용이면 NULL
+    requests = Column(Integer, nullable=False)                                                    # 이 IP 로 보낸 요청 수
+    ok_count = Column(Integer, nullable=False, server_default=text("0"))                          # 성공 검색 수
+    blocked_count = Column(Integer, nullable=False, server_default=text("0"))                     # 차단 감지 수
+    elapsed_sec = Column(Integer, nullable=True)                                                  # 세션 지속 시간(초)
+    end_reason = Column(String(20), nullable=False)                                               # budget/block/proxy_error/window/idle/shutdown/rotate
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()"))    # 세션 종료 시각
+
+    __table_args__ = (
+        # 상한 튜닝 쿼리(소스·기간별 종료 사유 분포) 최적화
+        Index("ix_ip_session_source", "source", "created_at"),
+    )
+
+
 class bot_detection(MAIN_BASE):
     """봇 감지 이력 — '이 IP로 몇 번째 요청에서, 어떤 방식으로 차단됐나'를 축적해 패턴 분석.
     (예: SELECT avg(ip_request_no) → IP당 평균 몇 요청 만에 감지되는지)"""

lps/crud/ip_session.py

@@ -0,0 +1,41 @@
+"""IP 세션 종료 이력 CRUD — 세션당 요청 수·종료 사유를 축적(요청 예산 상한 튜닝용)."""
+
+from sqlalchemy import text
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import DBType, DBWRType
+
+
+class IpSessionLog:
+    DB = DBType.MAIN.value
+
+    async def record(self, event: dict):
+        """세션 종료 이벤트 1건 저장. 기록 실패가 검색을 막지 않도록 호출부에서 예외를 삼킨다."""
+        sql = text("""
+            INSERT INTO ip_session (source, proxy_port, requests, ok_count, blocked_count, elapsed_sec, end_reason)
+            VALUES (:source, :proxy_port, :requests, :ok_count, :blocked_count, :elapsed_sec, :end_reason)
+        """)
+        params = {k: event.get(k) for k in
+                  ("source", "proxy_port", "requests", "ok_count", "blocked_count", "elapsed_sec", "end_reason")}
+        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 recent_stats(self, minutes: int = 60) -> dict:
+        """최근 N분 세션 요약 — 종료 사유별 건수(모니터링·알림용). 예: {"budget": 12, "block": 1}"""
+        sql = text("""
+            SELECT end_reason, count(*) FROM ip_session
+             WHERE created_at > now() - make_interval(mins => :m) GROUP BY end_reason
+        """)
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            rows = (await s.execute(sql, {"m": minutes})).all()
+            return {r[0]: int(r[1]) for r in rows}
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)

lps/docs/architecture.md

@@ -78,6 +78,7 @@
 
 **핵심 메커니즘**
 - **IP 회전(DECODO)**: 같은 IP로 계속 두드리면 차단 → 시간창 기반 sticky + 봇감지/전송오류 시 즉시 회전. 감지 이력(`bot_detection`)을 기록해 패턴 분석. **프록시 전송오류(407/터널)** 도 사이트 차단과 구분해 회전.
+- **선제 회전(요청 예산)**: IP당 요청 수가 예산(`LPS_IP_REQUEST_BUDGET`, 기본 3 — 실측상 5회 부근 차단)에 닿으면 **차단당하기 전에** 회전. 선제 교체된 포트는 평판이 깨끗해 로테이션 복귀 시 재사용됩니다. 반면 **차단 감지된 포트는 쿨다운**(`LPS_PORT_COOLDOWN_SEC`, 기본 max(sticky, 30분)) 동안 격리 — sticky 만료 후 복귀라 사실상 새 IP. 세션마다 `ip_session`(요청 수·종료 사유)을 남겨 예산 상한을 데이터로 튜닝합니다(쿼리는 database.md).
 - **시작 프리플라이트 + 웜업**: 기동 시 살아있는 프록시 포트를 선점(egress IP 로그)하고, 챌린지 소스를 미리 1회 풀어 **쿠키를 선점**(나쁜 IP는 회전 재시도) → 실 작업은 웜(빠름).
 - **동적 리소스 차단**: 이미지·폰트 등을 차단해 대역폭↓. 단 **Turnstile은 리소스 차단을 봇 신호로 감지**하므로, ESM은 챌린지 solving 중(콜드)엔 차단을 풀고 **cf_clearance 확보 후(웜)에만 차단**합니다.
 - **폴백 데드라인**: 오픈마켓 크롤은 '보강'이라 각 크롤에 시간 상한(기본 15초)을 둬, 한 몰이 안 풀려도 전체 지연이 늘지 않게 합니다.

lps/docs/database.md

@@ -14,6 +14,7 @@
 | `price_history` | 최저가 이력 — 그래프용 시계열 스냅샷 |
 | `search_negative` | 네거티브 캐시 — "없음"으로 확인된 상품을 일정 시간 기억 |
 | `bot_detection` | 봇 감지 이력 — 쿠팡이 차단한 패턴 기록 |
+| `ip_session` | IP 세션 종료 이력 — 요청 예산(선제 회전) 상한 튜닝 데이터 |
 
 ---
 
@@ -102,6 +103,33 @@ SELECT avg(ip_request_no), count(*) FROM bot_detection;
 
 ---
 
+## 5. `ip_session` — IP(프록시 포트) 세션 종료 이력
+
+브라우저(=IP 세션)가 끝날 때마다 기록. `bot_detection`은 **차단된** 세션만 남지만,
+여기엔 **무사 종료**(예산 선제 회전·시간창 만료 등)도 남아 요청 예산(`LPS_IP_REQUEST_BUDGET`)
+상한 튜닝의 원천 데이터가 됩니다.
+
+| 컬럼 | 뜻 |
+|------|-----|
+| `source` | 소스(coupang 등) |
+| `proxy_port` | 사용 포트(=IP 세션). 프록시 미사용이면 NULL |
+| `requests` | 이 IP로 보낸 요청 수 |
+| `ok_count` / `blocked_count` | 성공 검색 수 / 차단 감지 수 |
+| `elapsed_sec` | 세션 지속 시간(초) |
+| `end_reason` | 종료 사유 — `budget`(예산 선제) / `block`(차단) / `proxy_error`(포트 사망) / `window`(시간창 만료) / `idle`(유휴 정리) / `shutdown`(종료) |
+| `created_at` | 세션 종료 시각 |
+
+**예산 튜닝 쿼리** — 차단이 나기 시작하는 요청 수 분포를 보고 상한을 조정:
+```sql
+-- 종료 사유별 분포(최근 7일): budget 이 대다수 + block 0 이면 예산을 1씩 올려볼 수 있고,
+-- block 이 보이면 그 세션들의 requests 최솟값보다 예산을 낮게 유지한다.
+SELECT end_reason, count(*), avg(requests)::numeric(5,1) AS avg_req, min(requests), max(requests)
+  FROM ip_session WHERE created_at > now() - interval '7 days'
+ GROUP BY end_reason ORDER BY count(*) DESC;
+```
+
+---
+
 ## 스키마 생성/관리
 
 - 개발·테스트: SQLAlchemy 모델에서 `create_all`로 자동 생성.

lps/docs/operations.md

@@ -191,6 +191,9 @@ docker ps                          # lps-worker "(healthy)" 확인
 - **워커 헬스**: HEALTHCHECK(하트비트<120s)로 행 워커 감지. compose 의 `restart` 는 unhealthy 를
   재시작하지 않으므로 **autoheal 컨테이너**(라벨 `autoheal=true` 감시)가 재시작 담당. k8s 는 liveness probe 로 대체.
 - **잡 데드라인**: 잡 1건 300s 상한(`LPS_JOB_DEADLINE_SEC`) — 크롤 행이 워커 슬롯을 영구 점유하지 못하게 함.
+- **IP 선제 회전**: `LPS_IP_REQUEST_BUDGET`(기본 3) — IP당 요청 예산, 도달 시 차단 전에 회전(0=비활성).
+  `LPS_PORT_COOLDOWN_SEC`(기본 max(sticky, 1800)) — 차단 감지된 포트 격리 시간. 포트 수를 늘리면
+  (DECODO_PORT_START/END) 자동 반영 — 코드에 포트 수 하드코딩 없음. 튜닝은 `ip_session` 분석 쿼리(database.md) 참고.
 
 **남은 배포 과제**: API 인증·레이트리밋(비용 남용 방지), 다중 레플리카 시 분산 레이트리밋/프록시 IP 조정.
 **비용**: 대역폭이 원가의 대부분(오픈마켓 크롤) — 같은 상품 재크롤을 줄이는 **TTL 캐시**가 다음 절감 후보.

lps/migrations/2026-07-13-ip_session.sql

@@ -0,0 +1,17 @@
+-- IP(프록시 포트) 세션 종료 이력 — 요청 예산(LPS_IP_REQUEST_BUDGET) 상한 튜닝의 원천 데이터.
+-- bot_detection 은 차단된 세션만 남지만, 여기엔 무사 종료(budget/window 등)도 남는다.
+-- 적용: psql -h <host> -U <user> -d lps_db -f migrations/2026-07-13-ip_session.sql
+-- (postgres-init/init-data/init.sql 의 lps_db 섹션에도 동일 DDL 반영됨 — 신규 설치는 그쪽이 소스)
+
+CREATE TABLE IF NOT EXISTS ip_session (
+    id            uuid         PRIMARY KEY DEFAULT gen_random_uuid(),
+    source        VARCHAR(20)  NOT NULL,                                  -- coupang 등
+    proxy_port    INTEGER      NULL,                                      -- 사용 포트(=IP 세션), 프록시 미사용이면 NULL
+    requests      INTEGER      NOT NULL,                                  -- 이 IP 로 보낸 요청 수
+    ok_count      INTEGER      NOT NULL DEFAULT 0,                        -- 성공 검색 수
+    blocked_count INTEGER      NOT NULL DEFAULT 0,                        -- 차단 감지 수
+    elapsed_sec   INTEGER      NULL,                                      -- 세션 지속 시간(초)
+    end_reason    VARCHAR(20)  NOT NULL,                                  -- budget/block/proxy_error/window/idle/shutdown/rotate
+    created_at    TIMESTAMPTZ  NOT NULL DEFAULT now()                     -- 세션 종료 시각
+);
+CREATE INDEX IF NOT EXISTS ix_ip_session_source ON ip_session (source, created_at);

lps/services/search/browser_base.py

@@ -33,6 +33,10 @@ _BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"}
 _CHROME_CHANNEL = os.environ.get("LPS_CHROME_CHANNEL", "chrome")
 _CHROME_EXECUTABLE = os.environ.get("LPS_CHROME_EXECUTABLE") or None
 
+# IP(포트 세션)당 요청 예산 — 도달하면 차단당하기 **전에** 선제 회전해 IP 평판을 보존한다.
+# 실측상 5회 부근에서 차단된 이력이 있어 보수적으로 3회. 0 이면 비활성(시간창 회전만).
+_IP_REQUEST_BUDGET = int(os.environ.get("LPS_IP_REQUEST_BUDGET", "3"))
+
 
 def detect_block(html: str, product_count: int, markers: tuple, min_len: int) -> str | None:
     """0건 응답의 차단 여부 판정(순수 함수). 반환: 차단 마커(차단) 또는 None(정상 빈결과).
@@ -73,7 +77,8 @@ class BrowserSearchAdapter(SearchAdapter):
     max_proxy_retries: int = 2                   # 프록시 전송오류(포트 사망) 시 IP 회전 재시도 횟수
 
     def __init__(self, headless: bool = False, user_data_dir: str | None = None, rate_limiter: RateLimiter | None = None,
-                 proxy=None, block_resources: bool | None = None, on_detect=None, max_block_retries: int = 1):
+                 proxy=None, block_resources: bool | None = None, on_detect=None, max_block_retries: int = 1,
+                 ip_request_budget: int | None = None, on_session_end=None):
         self._headless = headless
         self._user_data_dir = user_data_dir or f"/tmp/lps_{self.source}_profile"
         self._rl = rate_limiter or RateLimiter()
@@ -82,6 +87,8 @@ class BrowserSearchAdapter(SearchAdapter):
         self._block_active = self._block_resources   # 요청별 실제 차단 여부(_blocking_now 로 갱신)
         self._on_detect = on_detect             # async def(event: dict) — 감지 영속화(선택)
         self._max_block_retries = max_block_retries
+        self._ip_budget = _IP_REQUEST_BUDGET if ip_request_budget is None else ip_request_budget
+        self._on_session_end = on_session_end   # async def(event: dict) — IP 세션 종료 기록(선택, 상한 튜닝 데이터)
         self._pw = None
         self._ctx = None
         self._launched_at = 0.0
@@ -91,6 +98,9 @@ class BrowserSearchAdapter(SearchAdapter):
         self._lock = asyncio.Lock()
         self._ok = 0
         self._blocked = 0
+        self._sess_ok = 0                       # 현재 IP 세션의 성공/차단(세션 종료 기록용, 재기동 시 리셋)
+        self._sess_blocked = 0
+        self._end_reason = None                 # 이번 세션이 끝나는 이유(budget/block/proxy_error/window/idle/shutdown)
         self._last_used = 0.0   # 마지막 검색 시각(monotonic) — 유휴 브라우저 정리 판단용
         self._cdp = None        # CDP 세션(실제 네트워크 바이트 계측용). 미지원 시 None → DOM 크기 폴백
         self._net_bytes = 0     # 현재 검색의 실제 전송 바이트(encodedDataLength 누적)
@@ -143,6 +153,8 @@ class BrowserSearchAdapter(SearchAdapter):
         if self._ctx is not None:
             if self._recycle_due():
                 LOG.d(f"[{self.source}] 브라우저 재기동(IP 회전)")
+                if self._end_reason is None:   # force 가 아닌 시간창 만료 재기동
+                    self._end_reason = "window"
                 await self._close_ctx()
             else:
                 return
@@ -163,16 +175,35 @@ class BrowserSearchAdapter(SearchAdapter):
             await self._ctx.route("**/*", self._route)
         self._launched_at = time.monotonic()
         self._ip_requests = 0
+        self._sess_ok = self._sess_blocked = 0
+        self._end_reason = None
         self._force_recycle = False
 
     async def _close_ctx(self):
         self._cdp = None   # 컨텍스트와 함께 CDP 세션도 죽음 → 다음 검색 때 재부착
         if self._ctx is not None:
+            await self._record_session_end()
             try:
                 await self._ctx.close()
             finally:
                 self._ctx = None
 
+    async def _record_session_end(self):
+        """IP 세션 종료 1건 기록 — '이 IP 로 몇 번 요청하고 어떻게 끝났나'. 예산(상한) 튜닝의 원천 데이터.
+        요청이 없던 세션(유휴 정리 등)은 노이즈라 기록하지 않는다. 기록 실패가 검색을 막지 않는다."""
+        if self._on_session_end is None or self._ip_requests == 0:
+            self._end_reason = None
+            return
+        event = {"source": self.source, "proxy_port": self._current_port,
+                 "requests": self._ip_requests, "ok_count": self._sess_ok, "blocked_count": self._sess_blocked,
+                 "elapsed_sec": int(time.monotonic() - self._launched_at),
+                 "end_reason": self._end_reason or "window"}
+        self._end_reason = None
+        try:
+            await self._on_session_end(event)
+        except Exception as ex:
+            LOG.e_no_callstack(f"[{self.source}] IP 세션 기록 실패(무시): {ex}")
+
     def _add_net(self, event):
         """CDP Network.loadingFinished 콜백 — 실제 전송 바이트(encodedDataLength) 누적."""
         try:
@@ -196,12 +227,18 @@ class BrowserSearchAdapter(SearchAdapter):
         """이 어댑터가 프록시(DECODO)를 경유하는지 — 대역폭 비용 귀속용."""
         return bool(self._proxy and self._proxy.enabled)
 
-    def _rotate_ip(self, reason: str):
+    def _budget_reached(self) -> bool:
-        """즉시 다음 IP(포트)로 회전 예약 + 다음 _ensure_browser 에서 브라우저 재기동."""
+        """현재 IP 로 요청 예산을 소진했는지(선제 회전 트리거). 프록시 미사용·예산 0(비활성)이면 False."""
+        return self.uses_proxy and self._ip_budget > 0 and self._ip_requests >= self._ip_budget
+
+    def _rotate_ip(self, reason: str, kind: str = "rotate", warn: bool = True):
+        """즉시 다음 IP(포트)로 회전 예약 + 다음 _ensure_browser 에서 브라우저 재기동.
+        kind 는 세션 종료 사유로 기록된다(budget=선제/block=차단/proxy_error=포트사망)."""
         if self._proxy and self._proxy.enabled:
             self._proxy.rotate()
             self._force_recycle = True
-            LOG.w(f"[{self.source}] IP 회전 — {reason}")
+            self._end_reason = kind
+            (LOG.w if warn else LOG.i)(f"[{self.source}] IP 회전 — {reason}")
 
     # ---- 검색(프록시 전송오류·봇 감지 → IP 회전 인라인 재시도) --------
     async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]:
@@ -210,6 +247,10 @@ class BrowserSearchAdapter(SearchAdapter):
             proxy_retries, block_retries = self.max_proxy_retries, self._max_block_retries
             while True:
                 await self._rl.wait()
+                # 예산 도달 → 차단당하기 전에 선제 회전. 이 포트는 불탄 게 아니라 쿨다운 없이
+                # 로테이션 복귀 시 재사용된다(IP 평판 보존이 목적).
+                if self._budget_reached():
+                    self._rotate_ip(f"요청예산 {self._ip_budget}회 도달 — 선제 회전", kind="budget", warn=False)
                 await self._ensure_browser()
                 self._ip_requests += 1
                 page = self._ctx.pages[0] if self._ctx.pages else await self._ctx.new_page()
@@ -225,7 +266,8 @@ class BrowserSearchAdapter(SearchAdapter):
                     # 프록시 전송 실패(포트/IP 사망·407)면 사이트 문제가 아니므로 IP 회전 후 재시도
                     if self.uses_proxy and is_proxy_error(str(ex)) and proxy_retries > 0:
                         proxy_retries -= 1
-                        self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}")
+                        self._proxy.mark_burned(self._current_port)   # 죽은 포트 — 쿨다운 뒤 복귀(sticky 만료로 새 IP)
+                        self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}", kind="proxy_error")
                         continue
                     raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex
 
@@ -234,6 +276,7 @@ class BrowserSearchAdapter(SearchAdapter):
                 products = self._parse(html)
                 if products:
                     self._ok += 1
+                    self._sess_ok += 1
                     LOG.d(f"[{self.source}] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})")
                     return products[:limit]
 
@@ -241,13 +284,18 @@ class BrowserSearchAdapter(SearchAdapter):
                 blocked = marker is not None
                 self._blocked += 1
                 if blocked:
+                    self._sess_blocked += 1
                     await self._report_detection(query, marker, len(html))
+                    if self.uses_proxy:
+                        self._proxy.mark_burned(self._current_port)   # 불탄 포트 — 쿨다운 격리(로테이션이 건너뜀)
 
                 if blocked and self.uses_proxy and block_retries > 0:
                     block_retries -= 1
-                    self._rotate_ip(f"봇 감지 재시도 {self._max_block_retries - block_retries}/{self._max_block_retries}")
+                    self._rotate_ip(f"봇 감지 재시도 {self._max_block_retries - block_retries}/{self._max_block_retries}", kind="block")
                     continue
 
+                if blocked:   # 재시도 소진/비활성 — 불탄 포트로 다음 검색을 하지 않도록 회전만 예약하고 포기
+                    self._rotate_ip("봇 감지 — 다음 검색은 새 IP", kind="block")
                 raise AdapterError(f"{self.source} 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked)
 
     async def _report_detection(self, query: str, marker: str, html_len: int):
@@ -279,12 +327,14 @@ class BrowserSearchAdapter(SearchAdapter):
         async with self._lock:
             if self._ctx is not None and time.monotonic() - self._last_used >= idle_sec:
                 LOG.d(f"[{self.source}] 유휴 {idle_sec:.0f}s 초과 → 브라우저 정리(다음 검색 때 재기동)")
+                if self._end_reason is None:
+                    self._end_reason = "idle"
                 await self._close_ctx()
 
     async def close(self):
-        if self._ctx is not None:
+        if self._end_reason is None:
-            await self._ctx.close()
+            self._end_reason = "shutdown"
-            self._ctx = None
+        await self._close_ctx()
         if self._pw is not None:
             await self._pw.stop()
             self._pw = None

lps/services/search/proxy.py

@@ -9,6 +9,7 @@ Decodo residential 은 **포트 기반 sticky** 모델이다:
 자격증명/엔드포인트는 config.local.toml [DecodoConfig] 에서 로드(시크릿).
 """
 
+import os
 import time
 from urllib.parse import quote
 
@@ -28,24 +29,52 @@ class DecodoProxy:
         self.port_end = cfg.port_end
         self.session_minutes = cfg.session_minutes or 10
         self._rotate_offset = 0  # 봇 감지 등으로 '즉시 회전'이 필요할 때 증가
+        # 불탄(차단 감지된) 포트 격리 시간. sticky 만료(session_minutes) 이상이어야
+        # 쿨다운 복귀 시 같은 포트라도 사실상 새 IP 가 배정된다. 기본 max(sticky, 30분).
+        self.cooldown_sec = int(os.environ.get("LPS_PORT_COOLDOWN_SEC", "0")) \
+            or max(self.session_minutes * 60, 1800)
+        self._burned: dict[int, float] = {}   # port → 쿨다운 만료 시각(monotonic)
 
     @property
     def enabled(self) -> bool:
         return all([self.host, self.username, self.password, self.port_start, self.port_end])
 
     def rotate(self):
-        """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지 시 호출."""
+        """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지·예산 도달 시 호출."""
         self._rotate_offset += 1
 
     def seed_offset(self, k: int):
         """워커별 시작 포트 분산용 — 동시 워커가 같은 포트(=같은 IP)를 쓰지 않도록 시작점을 벌린다."""
         self._rotate_offset = k
 
+    def mark_burned(self, port: int | None, cooldown_sec: float | None = None):
+        """차단 감지된 포트를 쿨다운 격리 — _port() 가 만료 전까지 건너뛴다.
+        선제(예산) 회전된 포트는 부르지 않는다 — 불탄 게 아니므로 로테이션 복귀 시 재사용."""
+        if port is None:
+            return
+        self._burned[port] = time.monotonic() + (cooldown_sec if cooldown_sec is not None else self.cooldown_sec)
+        LOG.i(f"[proxy] 포트 {port} 쿨다운 {int(cooldown_sec or self.cooldown_sec)}s — 활성 {self.available_ports()}/{self.port_end - self.port_start + 1}")
+
+    def available_ports(self) -> int:
+        """쿨다운 중이 아닌 포트 수(관측·알림용)."""
+        self._prune_burned()
+        return (self.port_end - self.port_start + 1) - len(self._burned)
+
+    def _prune_burned(self):
+        now = time.monotonic()
+        self._burned = {p: t for p, t in self._burned.items() if t > now}
+
     def _port(self) -> int:
-        """시간창 + 수동 오프셋 기반 포트 선택. 창 안에선 동일 IP, rotate()나 창 변화 시 다음 IP."""
+        """시간창 + 수동 오프셋 기반 포트 선택. 창 안에선 동일 IP, rotate()나 창 변화 시 다음 IP.
+        쿨다운 중인 포트는 건너뛰고, 전 포트가 쿨다운이면 만료가 가장 임박한 포트를 쓴다(가용성 우선)."""
         n = self.port_end - self.port_start + 1
         bucket = int(time.time() // (self.session_minutes * 60))
-        return self.port_start + ((bucket + self._rotate_offset) % n)
+        self._prune_burned()
+        for k in range(n):
+            port = self.port_start + ((bucket + self._rotate_offset + k) % n)
+            if port not in self._burned:
+                return port
+        return min(self._burned, key=self._burned.get)
 
     @property
     def current_port(self):

lps/tests/test_browser_base.py

@@ -49,8 +49,9 @@ class _MockCtx:
 class _MockProxy:
     enabled = True
     session_minutes = 10
-    def __init__(self): self.rotations = 0
+    def __init__(self): self.rotations, self.burned = 0, []
     def rotate(self): self.rotations += 1
+    def mark_burned(self, port, cooldown_sec=None): self.burned.append(port)
     def playwright_proxy(self): return None
     @property
     def current_port(self): return 10001
@@ -70,6 +71,7 @@ class _MockAdapter(BrowserSearchAdapter):
             self._ctx = _MockCtx(self._page)
             self._force_recycle = False
             self._ip_requests = 0
+            self._current_port = self._proxy.current_port if self._proxy else None  # 실제 _ensure_browser 와 동일
 
     async def _ensure_net_meter(self, page): pass   # CDP 없음 → last_bytes=DOM 크기
     async def _wait_ready(self, page): pass
@@ -110,11 +112,12 @@ async def test_non_proxy_goto_error_raises_no_rotation():
 
 
 async def test_persistent_block_exhausts_and_raises_blocked():
-    # 차단 2회 연속(max_block_retries=1) → 회전 1회 후 소진 → blocked=True 로 실패
+    # 차단 2회 연속(max_block_retries=1) → 재시도 회전 1회 + 소진 후 '다음 검색용' 회전 1회 → blocked=True 로 실패
     ad = _ad(["x", "x"])
     with pytest.raises(AdapterError) as ei:
         await ad.search("q")
-    assert ei.value.blocked is True and ad._proxy.rotations == 1
+    assert ei.value.blocked is True and ad._proxy.rotations == 2
+    assert ad._proxy.burned == [10001, 10001]   # 차단마다 해당 포트 쿨다운 격리
 
 
 # ── 라이브 스모크(옵트인): 실제 사이트 셀렉터·안티봇 드리프트 감지 ──────

lps/tests/test_ip_session.py

@@ -0,0 +1,136 @@
+"""IP 선제 로테이션 테스트 — 요청 예산 판정·세션 종료 기록(브라우저 불필요) + CRUD(실 lps_db)."""
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from config.config_models import DecodoConfig
+from crud.ip_session import IpSessionLog
+from services.search.coupang.adapter import CoupangAdapter
+from services.search.proxy import DecodoProxy
+
+
+def _proxy():
+    return DecodoProxy(DecodoConfig(host="gate.decodo.com", username="u", password="p",
+                                    port_start=10001, port_end=10010, session_minutes=10))
+
+
+def _adapter(**kw):
+    # __init__ 은 브라우저를 띄우지 않는다 — 예산/세션 부기 로직만 검증
+    return CoupangAdapter(headless=True, user_data_dir="/tmp/lps_test_profile", **kw)
+
+
+# ---- 요청 예산(선제 회전) 판정 ------------------------------------------
+
+def test_budget_reached_only_with_proxy():
+    ad = _adapter(ip_request_budget=3)
+    ad._ip_requests = 3
+    assert ad._budget_reached() is False       # 프록시 미사용 — 예산 개념 없음
+    ad._proxy = _proxy()
+    assert ad._budget_reached() is True
+
+
+def test_budget_zero_disables_preemptive_rotation():
+    ad = _adapter(ip_request_budget=0, proxy=_proxy())
+    ad._ip_requests = 999
+    assert ad._budget_reached() is False
+
+
+def test_budget_not_reached_below_threshold():
+    ad = _adapter(ip_request_budget=3, proxy=_proxy())
+    ad._ip_requests = 2
+    assert ad._budget_reached() is False
+
+
+def test_default_budget_is_conservative():
+    ad = _adapter(proxy=_proxy())
+    assert ad._ip_budget == 3                  # 실측상 5회 부근 차단 → 기본 3회
+
+
+def test_rotate_ip_records_end_reason_and_recycles():
+    ad = _adapter(proxy=_proxy())
+    before = ad._proxy.current_port
+    ad._rotate_ip("예산 도달", kind="budget", warn=False)
+    assert ad._end_reason == "budget"
+    assert ad._force_recycle is True
+    assert ad._proxy.current_port != before    # 즉시 다음 포트
+
+
+# ---- 세션 종료 기록 -------------------------------------------------------
+
+async def test_session_end_event_shape():
+    events = []
+
+    async def cb(e):
+        events.append(e)
+
+    ad = _adapter(proxy=_proxy(), on_session_end=cb)
+    ad._ip_requests, ad._sess_ok, ad._sess_blocked = 3, 2, 1
+    ad._current_port, ad._end_reason = 10005, "budget"
+    await ad._record_session_end()
+    assert events == [{
+        "source": "coupang", "proxy_port": 10005, "requests": 3,
+        "ok_count": 2, "blocked_count": 1, "elapsed_sec": events[0]["elapsed_sec"],
+        "end_reason": "budget",
+    }]
+    assert ad._end_reason is None              # 기록 후 리셋
+
+
+async def test_session_end_skips_empty_session():
+    events = []
+
+    async def cb(e):
+        events.append(e)
+
+    ad = _adapter(on_session_end=cb)
+    ad._ip_requests = 0                        # 요청 없던 세션(유휴 정리 등)은 노이즈 — 미기록
+    await ad._record_session_end()
+    assert events == []
+
+
+async def test_session_end_defaults_to_window_reason():
+    events = []
+
+    async def cb(e):
+        events.append(e)
+
+    ad = _adapter(on_session_end=cb)
+    ad._ip_requests = 2                        # end_reason 미지정 = 시간창 만료 재기동
+    await ad._record_session_end()
+    assert events[0]["end_reason"] == "window"
+
+
+async def test_session_end_callback_failure_is_swallowed():
+    async def boom(e):
+        raise RuntimeError("db down")
+
+    ad = _adapter(on_session_end=boom)
+    ad._ip_requests = 1
+    await ad._record_session_end()             # 예외가 검색을 막지 않는다
+
+
+# ---- CRUD (실 lps_db) ----------------------------------------------------
+
+@pytest_asyncio.fixture
+async def ip_log(db_engine):
+    async with db_engine.begin() as conn:
+        await conn.execute(text("TRUNCATE ip_session"))
+    return IpSessionLog()
+
+
+async def test_record_persists_session(ip_log, db_engine):
+    await ip_log.record({"source": "coupang", "proxy_port": 10003, "requests": 3,
+                         "ok_count": 3, "blocked_count": 0, "elapsed_sec": 120, "end_reason": "budget"})
+    async with db_engine.begin() as conn:
+        row = (await conn.execute(text(
+            "SELECT source, proxy_port, requests, ok_count, end_reason FROM ip_session"
+        ))).first()
+    assert row.source == "coupang" and row.proxy_port == 10003
+    assert row.requests == 3 and row.ok_count == 3 and row.end_reason == "budget"
+
+
+async def test_recent_stats_groups_by_reason(ip_log, db_engine):
+    for reason in ("budget", "budget", "block"):
+        await ip_log.record({"source": "coupang", "requests": 3, "ok_count": 3,
+                             "blocked_count": 0, "end_reason": reason})
+    stats = await ip_log.recent_stats(60)
+    assert stats == {"budget": 2, "block": 1}

lps/tests/test_proxy.py

@@ -50,3 +50,38 @@ def test_rotate_advances_port_immediately():
     assert 10001 <= after <= 10003
     p.rotate(); p.rotate()                      # 3번 회전하면 한 바퀴 → 원위치
     assert p._port() == before
+
+
+def test_burned_port_is_skipped():
+    p = _p(port_start=10001, port_end=10003)
+    burned = p._port()
+    p.mark_burned(burned)
+    assert p._port() != burned                 # 쿨다운 중인 포트는 건너뜀
+    assert p.available_ports() == 2
+
+
+def test_burned_port_returns_after_cooldown():
+    p = _p(port_start=10001, port_end=10003)
+    burned = p._port()
+    p.mark_burned(burned, cooldown_sec=0)      # 즉시 만료
+    assert p._port() == burned                 # 만료 후엔 다시 사용 가능
+    assert p.available_ports() == 3
+
+
+def test_all_burned_falls_back_to_earliest_expiry():
+    p = _p(port_start=10001, port_end=10002)   # 포트 2개
+    first, second = 10001, 10002
+    p.mark_burned(first, cooldown_sec=10)      # 먼저 만료
+    p.mark_burned(second, cooldown_sec=999)
+    assert p._port() == first                  # 전 포트 쿨다운 — 만료 임박 포트 사용(가용성 우선)
+
+
+def test_mark_burned_ignores_none_port():
+    p = _p()
+    p.mark_burned(None)                        # 프록시 미사용 세션 — no-op
+    assert p.available_ports() == 10
+
+
+def test_cooldown_default_at_least_30min():
+    assert _p(session_minutes=10).cooldown_sec == 1800          # max(sticky 600, 1800)
+    assert _p(session_minutes=60).cooldown_sec == 3600          # sticky 가 더 길면 sticky 만큼

lps/worker_main.py

@@ -17,6 +17,7 @@ from config.server_configs import web_server_config, openai_config, decodo_confi
 from crud.job_crud import JobQueue
 from crud.negative_cache import NegativeCache
 from crud.bot_detection import BotDetectionLog
+from crud.ip_session import IpSessionLog
 from crud.price_history import PriceHistory
 from services.search.proxy import DecodoProxy
 from services.search.coupang.adapter import CoupangAdapter
@@ -56,6 +57,7 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history
         n = proxy.port_end - proxy.port_start + 1
         proxy.seed_offset(i * max(1, n // concurrency))
     bot_log = BotDetectionLog()
+    ip_log = IpSessionLog()
     suffix = f"_w{i}" if concurrency > 1 else ""
 
     def _pf(source):  # 워커별 Chrome 프로필 경로(중복 실행 시 ProcessSingleton 충돌 방지)
@@ -64,7 +66,8 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history
         return f"{base}/lps_{source}{suffix}"
 
     adapters = {
-        "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy, on_detect=bot_log.record),
+        "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy,
+                                  on_detect=bot_log.record, on_session_end=ip_log.record),
         "naver": NaverAdapter(),   # httpx 직접(프록시 미경유) — 워커별 인스턴스(last_bytes 경합 회피)
     }
     # 폴백은 기본 비활성(LPS_FALLBACKS 로 켬 — 상단 주석 참고). 켤 땐 데드라인이 상한이라
@@ -72,9 +75,11 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history
     fallback_adapters = {}
     for name in _enabled_fallbacks():
         if name == "st11":
-            fallback_adapters[name] = ElevenStAdapter(headless=False, user_data_dir=_pf("st11"), proxy=proxy, on_detect=bot_log.record, max_block_retries=0)
+            fallback_adapters[name] = ElevenStAdapter(headless=False, user_data_dir=_pf("st11"), proxy=proxy,
+                                                      on_detect=bot_log.record, on_session_end=ip_log.record, max_block_retries=0)
         else:
-            fallback_adapters[name] = EsmAdapter(name, headless=False, user_data_dir=_pf(name), proxy=proxy, on_detect=bot_log.record, max_block_retries=0)
+            fallback_adapters[name] = EsmAdapter(name, headless=False, user_data_dir=_pf(name), proxy=proxy,
+                                                 on_detect=bot_log.record, on_session_end=ip_log.record, max_block_retries=0)
     # AI 도 워커별 인스턴스 — 공유 상태(last_usage) 경합 원천 제거
     judge = SimilarityJudge() if has_openai else None
     keyword_gen = KeywordGenerator() if has_openai else None

postgres-init/init-data/init.sql

@@ -744,6 +744,20 @@ CREATE TABLE IF NOT EXISTS price_history (
 );
 CREATE INDEX IF NOT EXISTS ix_price_history_product ON price_history (product_code, triggered_at);
 
+-- IP(프록시 포트) 세션 종료 이력 — 요청 예산(LPS_IP_REQUEST_BUDGET) 상한 튜닝의 원천 데이터
+CREATE TABLE IF NOT EXISTS ip_session (
+    id            uuid         PRIMARY KEY DEFAULT gen_random_uuid(),
+    source        VARCHAR(20)  NOT NULL,                                  -- coupang 등
+    proxy_port    INTEGER      NULL,                                      -- 사용 포트(=IP 세션), 프록시 미사용이면 NULL
+    requests      INTEGER      NOT NULL,                                  -- 이 IP 로 보낸 요청 수
+    ok_count      INTEGER      NOT NULL DEFAULT 0,                        -- 성공 검색 수
+    blocked_count INTEGER      NOT NULL DEFAULT 0,                        -- 차단 감지 수
+    elapsed_sec   INTEGER      NULL,                                      -- 세션 지속 시간(초)
+    end_reason    VARCHAR(20)  NOT NULL,                                  -- budget/block/proxy_error/window/idle/shutdown/rotate
+    created_at    TIMESTAMPTZ  NOT NULL DEFAULT now()                     -- 세션 종료 시각
+);
+CREATE INDEX IF NOT EXISTS ix_ip_session_source ON ip_session (source, created_at);
+
 -- 봇 감지 이력 — IP·요청순번·차단방식 축적(패턴 분석)
 CREATE TABLE IF NOT EXISTS bot_detection (
     id            uuid         PRIMARY KEY DEFAULT gen_random_uuid(),