feat: Migrate batch analysis DB writes to db_worker queue pattern

backend/app/services/analysis_db_service.py

@@ -1,11 +1,17 @@
 """StockAnalysis DB 서비스 — 단일 진실 공급원 (Single Source of Truth)
 
 StockAnalysis ORM 저장 로직을 중앙화하여 3곳의 중복 제거:
-  - batch_daily_analysis.py   (bulk)
+  - batch_daily_analysis.py    (bulk)
   - batch_intraday_analysis.py (per-stock, rescore)
 
 PostgreSQL UPSERT(ON CONFLICT DO UPDATE)를 사용하여
-DELETE → INSERT 사이의 race condition을 원자적으로 해결.
+DELETE → INSERT 사이의 race condition 을 원자적으로 해결.
+
+워커 패턴 (CLAUDE.md 3.2):
+  - session 을 직접 받는 `upsert_analysis` / `bulk_upsert_analysis` 는
+    scheduler / batch 에서 기존처럼 사용 가능 (시그니처 불변).
+  - 실시간 콜백·인트라데이 분석에서는 `queue_upsert_analysis` /
+    `queue_bulk_upsert_analysis` 를 사용하여 DB 락 경합을 방지.
 """
 from __future__ import annotations
 
@@ -26,10 +32,13 @@ if TYPE_CHECKING:
     from app.api.endpoints.recommendations import StockRecommendation
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# 헬퍼
+# ─────────────────────────────────────────────────────────────────────────────
 def passes_price_filter(result: "StockRecommendation") -> bool:
-    """5,000원 이하 저가주 필터 (CLAUDE.md 5절).
+    """5,000원 이하 저가주 필터 (CLAUDE.md 5.4).
 
-    True → 통과 (거래 가능), False → 필터됨 (5,000원 미만).
+    True → 통과, False → 필터됨.
     """
     return not (result.current_price and result.current_price < 5000)
 
@@ -53,9 +62,9 @@ def to_analysis_dict(
     result: "StockRecommendation",
     analysis_date: date,
 ) -> dict:
-    """StockRecommendation → StockAnalysis INSERT 딕셔너리 변환.
+    """StockRecommendation → StockAnalysis INSERT dict 변환.
 
-    ORM 객체 대신 dict를 반환하여 단건 UPSERT와 bulk INSERT에서 모두 재사용.
+    ORM 객체 대신 dict 를 반환하여 단건 UPSERT 와 bulk INSERT 에서 모두 재사용.
     """
     return {
         "stock_code": stock_code,
@@ -84,6 +93,24 @@ def to_analysis_dict(
     }
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# 내부 실행 함수 (db_worker.py 에서도 호출)
+# ─────────────────────────────────────────────────────────────────────────────
+async def _execute_upsert(session: AsyncSession, rows: list[dict]) -> None:
+    """실제 PostgreSQL UPSERT 실행 — session 은 호출 측이 관리."""
+    if not rows:
+        return
+    stmt = pg_insert(StockAnalysis).values(rows)
+    stmt = stmt.on_conflict_do_update(
+        constraint="uq_stock_analysis_code_date",
+        set_={col: stmt.excluded[col] for col in _UPSERT_UPDATE_COLS},
+    )
+    await session.execute(stmt)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 기존 session 기반 함수 — 시그니처 불변 (CLAUDE.md 4.1)
+# ─────────────────────────────────────────────────────────────────────────────
 async def upsert_analysis(
     session: AsyncSession,
     stock_code: str,
@@ -91,19 +118,9 @@ async def upsert_analysis(
     result: "StockRecommendation",
     analysis_date: date,
 ) -> None:
-    """단건 UPSERT — 동일 (stock_code, analysis_date)가 있으면 UPDATE, 없으면 INSERT.
+    """단건 UPSERT — session 직접 사용. 배치 스케줄러에서 그대로 호출 가능."""
-
-    DELETE+INSERT 대비:
-      - 원자적 연산 (race condition 없음)
-      - 단일 DB 왕복 (네트워크 절반)
-    """
     row = to_analysis_dict(stock_code, stock_name, result, analysis_date)
-    stmt = pg_insert(StockAnalysis).values(**row)
+    await _execute_upsert(session, [row])
-    stmt = stmt.on_conflict_do_update(
-        constraint="uq_stock_analysis_code_date",
-        set_={col: stmt.excluded[col] for col in _UPSERT_UPDATE_COLS},
-    )
-    await session.execute(stmt)
 
 
 async def bulk_upsert_analysis(
@@ -111,35 +128,63 @@ async def bulk_upsert_analysis(
     items: list[tuple[str, str, "StockRecommendation"]],
     analysis_date: date,
 ) -> int:
-    """다건 UPSERT — 단일 INSERT ... ON CONFLICT 배치 실행.
+    """다건 UPSERT — session 직접 사용. 일봉 배치에서 그대로 호출 가능."""
-
-    items: [(stock_code, stock_name, StockRecommendation), ...]
-    Returns: 처리된 건수
-    """
     if not items:
         return 0
-
     rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items]
-    stmt = pg_insert(StockAnalysis).values(rows)
+    await _execute_upsert(session, rows)
-    stmt = stmt.on_conflict_do_update(
-        constraint="uq_stock_analysis_code_date",
-        set_={col: stmt.excluded[col] for col in _UPSERT_UPDATE_COLS},
-    )
-    await session.execute(stmt)
     return len(rows)
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# NEW: 워커 큐 기반 함수 — 실시간 콜백·인트라데이에서 사용 (CLAUDE.md 3.2)
+# ─────────────────────────────────────────────────────────────────────────────
+async def queue_upsert_analysis(
+    stock_code: str,
+    stock_name: str,
+    result: "StockRecommendation",
+    analysis_date: date,
+) -> None:
+    """단건 UPSERT 를 db_worker 큐에 위임.
+
+    실시간 스레드 / 인트라데이 분석에서 사용.
+    DB 세션을 직접 열지 않으므로 락 경합 없음.
+    """
+    from app.services.db_worker import enqueue_upsert_analysis
+    row = to_analysis_dict(stock_code, stock_name, result, analysis_date)
+    await enqueue_upsert_analysis(row)
+
+
+async def queue_bulk_upsert_analysis(
+    items: list[tuple[str, str, "StockRecommendation"]],
+    analysis_date: date,
+) -> int:
+    """다건 UPSERT 를 db_worker 큐에 위임.
+
+    배치 분석 결과를 비동기로 저장할 때 사용.
+    API 응답 지연 없이 큐에 넣고 즉시 반환.
+    Returns: 큐에 등록된 row 수
+    """
+    if not items:
+        return 0
+    from app.services.db_worker import enqueue_bulk_upsert_analysis
+    rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items]
+    await enqueue_bulk_upsert_analysis(rows)
+    return len(rows)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# DB 스캐너 — 저장된 분석 결과로 신호 종목 필터링 (API 호출 없음)
+# ─────────────────────────────────────────────────────────────────────────────
 async def run_db_scanner(analysis_date: date) -> set[str]:
-    """배치 DB 기반 스캐너 — KosdaqScanner와 동일한 기준을 단일일 데이터로 적용.
+    """배치 DB 기반 스캐너 — KosdaqScanner 와 동일 기준, 추가 API 호출 없음.
 
-    저장된 RSI / 볼린저 / 거래량 데이터만 사용하므로 추가 API 호출 없음.
+    조건:
-
-    조건 (성능 중심 필터):
       - RSI(14) ≤ 30  (과매도)
-      - 종가 ≤ BB 하단 × 1.015  (BB 하단 근처)
+      - 종가 ≤ BB 하단 × 1.015
       - 시세 ≥ 5,000원
-      - ADX ≤ 30  (지나치게 강한 추세는 평균회귀 부적합)
+      - ADX ≤ 30  (강한 추세 = 평균회귀 부적합)
-      - MA5 ≥ MA20 × 0.95  (ADX 방향 proxy — 강한 하락 추세 종목 제외)
+      - MA5 ≥ MA20 × 0.95  (강한 하락 추세 제외)
 
     Returns: 신호 종목 코드 set
     """
@@ -170,17 +215,14 @@ async def run_db_scanner(analysis_date: date) -> set[str]:
             if not bb_lower or bb_lower == 0:
                 continue
 
-            # BB 하단선 근처 확인 (-inf ~ +1.5%)
             pct = (price - bb_lower) / bb_lower * 100
             if pct > 1.5:
                 continue
 
-            # 구조적 추세장(ADX > 30)는 평균회귀에 부적합
             adx = row.adx_14
             if adx is not None and adx > 30:
                 continue
 
-            # ADX 방향 proxy: MA5 < MA20 × 0.95 → 강한 하락 추세 → 평균회귀 부적합
             ma5, ma20 = row.ma5, row.ma20
             if ma5 is not None and ma20 is not None and ma20 > 0:
                 if float(ma5) < float(ma20) * 0.95:
@@ -188,7 +230,10 @@ async def run_db_scanner(analysis_date: date) -> set[str]:
 
             signal_tickers.add(row.stock_code)
 
-        logger.info("🔍 DB 스캐너: %d개 신호 종목 (RSI≤30 + BB하단 + ADX≤30 + MA방향)", len(signal_tickers))
+        logger.info(
+            "🔍 DB 스캐너: %d개 신호 종목 (RSI≤30 + BB하단 + ADX≤30 + MA방향)",
+            len(signal_tickers),
+        )
         return signal_tickers
 
     except Exception as exc:

backend/app/services/db_worker.py

@@ -1,48 +1,102 @@
+"""Background DB Writer Worker
+
+ARCHITECTURE (CLAUDE.md 3.2):
+  - 실시간/콜백 스레드에서는 절대 DB 세션을 직접 열지 않는다.
+  - 모든 DB 쓰기 작업은 이 모듈의 db_task_queue 로 위임한다.
+  - 워커가 큐를 소비하며 단일 세션에서 순차 처리 → 락 경합 제거.
+
+Supported task types:
+  - insert_trade_history       : dict  → TradeHistory INSERT
+  - upsert_stock_analysis      : dict  → StockAnalysis single UPSERT
+  - bulk_upsert_stock_analysis : list  → StockAnalysis batch UPSERT
+"""
 import asyncio
 import logging
+from datetime import date
+
 from app.core.database import AsyncSessionLocal
-from app.models.models import TradeHistory, StockAnalysis
+from app.models.models import TradeHistory
 
 logger = logging.getLogger(__name__)
 
-# 전역(Global) 비동기 큐 생성 (메모리 버퍼 역할)
+# ─────────────────────────────────────────────────────────────────────────────
-db_task_queue = asyncio.Queue()
+# 전역 큐 — 앱 어디서나 import 하여 put() 호출
+# ─────────────────────────────────────────────────────────────────────────────
+db_task_queue: asyncio.Queue = asyncio.Queue()
 
-async def db_writer_worker():
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 편의 함수 — 호출 측에서 dict 구성 없이 바로 사용
+# ─────────────────────────────────────────────────────────────────────────────
+async def enqueue_trade_history(data: dict) -> None:
+    """매매일지 INSERT 를 큐에 등록."""
+    await db_task_queue.put({"type": "insert_trade_history", "data": data})
+
+
+async def enqueue_upsert_analysis(row: dict) -> None:
+    """단건 StockAnalysis UPSERT 를 큐에 등록."""
+    await db_task_queue.put({"type": "upsert_stock_analysis", "data": row})
+
+
+async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None:
+    """다건 StockAnalysis batch UPSERT 를 큐에 등록."""
+    if rows:
+        await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows})
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 워커 본체
+# ─────────────────────────────────────────────────────────────────────────────
+async def db_writer_worker() -> None:
+    """lifespan 에서 asyncio.create_task() 로 실행되는 영구 백그라운드 워커.
+
+    큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
+    commit / rollback 을 자동 관리한다.
     """
-    메인 스레드나 API 스레드를 막지 않고, 
+    logger.info("🛠️  Background DB Writer Worker started")
-    백그라운드에서 큐에 쌓인 DB 작업만 순차적으로 처리하는 워커
-    DB 커넥션 경합과 락(Lock)을 방지합니다.
-    """
-    logger.info("🛠️ Background DB Writer Worker started")
     while True:
+        task = await db_task_queue.get()
         try:
-            # 큐에 데이터가 들어올 때까지 대기
-            task = await db_task_queue.get()
             task_type = task.get("type")
             data = task.get("data")
 
-            # 워커 전용 DB 세션을 열어서 처리 (락 경합 최소화)
             async with AsyncSessionLocal() as session:
+                # ── 1. 매매일지 INSERT ─────────────────────────────────────
                 if task_type == "insert_trade_history":
-                    # 매매일지 추가
+                    session.add(TradeHistory(**data))
-                    new_trade = TradeHistory(**data)
-                    session.add(new_trade)
                     await session.commit()
-                    logger.info(f"✅ [Worker] 매매일지 비동기 저장 완료: {data.get('stock_code', 'Unknown')}")
+                    logger.info(
+                        "✅ [Worker] TradeHistory saved: %s",
+                        data.get("stock_code", "?"),
+                    )
 
+                # ── 2. 단건 StockAnalysis UPSERT ──────────────────────────
                 elif task_type == "upsert_stock_analysis":
-                    # 종목 분석 결과 저장 (추후 batch_analysis 등에서 이관될 로직용 플레이스홀더)
+                    from app.services.analysis_db_service import _execute_upsert
-                    # 실제 upsert 로직은 데이터 구조에 맞춰 구현
+                    await _execute_upsert(session, [data])
-                    logger.info(f"✅ [Worker] 종목분석 비동기 저장 처리 (Placeholder): {data.get('stock_code', 'Unknown')}")
+                    await session.commit()
+                    logger.debug(
+                        "✅ [Worker] StockAnalysis upsert: %s",
+                        data.get("stock_code", "?"),
+                    )
+
+                # ── 3. 다건 StockAnalysis batch UPSERT ────────────────────
+                elif task_type == "bulk_upsert_stock_analysis":
+                    from app.services.analysis_db_service import _execute_upsert
+                    await _execute_upsert(session, data)          # data = list[dict]
+                    await session.commit()
+                    logger.info(
+                        "✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data)
+                    )
 
                 else:
-                    logger.warning(f"⚠️ [Worker] 알 수 없는 작업 타입: {task_type}")
+                    logger.warning("⚠️  [Worker] Unknown task type: %s", task_type)
 
         except asyncio.CancelledError:
-            logger.info("⏹️ DB Writer Worker stopped")
+            logger.info("⏹️  DB Writer Worker stopped")
             break
-        except Exception as e:
+        except Exception as exc:
-            logger.error(f"❌ [Worker] DB 작업 중 에러 발생: {e}", exc_info=True)
+            logger.error("❌ [Worker] DB task failed: %s", exc, exc_info=True)
+            # 실패한 task 는 드롭 — 재시도가 필요하면 retry_count 추가 가능
         finally:
             db_task_queue.task_done()