From a2b076f26940a532f5e55af8ece9f32baa4d6fcf Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Tue, 24 Feb 2026 13:31:23 +0900 Subject: [PATCH] feat: Migrate batch analysis DB writes to db_worker queue pattern --- backend/app/services/analysis_db_service.py | 127 +++++++++++++------- backend/app/services/db_worker.py | 102 ++++++++++++---- 2 files changed, 164 insertions(+), 65 deletions(-) diff --git a/backend/app/services/analysis_db_service.py b/backend/app/services/analysis_db_service.py index 0c21cc9..2f9d3c9 100644 --- a/backend/app/services/analysis_db_service.py +++ b/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. - - DELETE+INSERT 대비: - - 원자적 연산 (race condition 없음) - - 단일 DB 왕복 (네트워크 절반) - """ + """단건 UPSERT — session 직접 사용. 배치 스케줄러에서 그대로 호출 가능.""" row = to_analysis_dict(stock_code, stock_name, result, analysis_date) - stmt = pg_insert(StockAnalysis).values(**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) + await _execute_upsert(session, [row]) 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 배치 실행. - - items: [(stock_code, stock_name, StockRecommendation), ...] - Returns: 처리된 건수 - """ + """다건 UPSERT — session 직접 사용. 일봉 배치에서 그대로 호출 가능.""" 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) - 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) + await _execute_upsert(session, rows) 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 (지나치게 강한 추세는 평균회귀 부적합) - - MA5 ≥ MA20 × 0.95 (ADX 방향 proxy — 강한 하락 추세 종목 제외) + - ADX ≤ 30 (강한 추세 = 평균회귀 부적합) + - 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: diff --git a/backend/app/services/db_worker.py b/backend/app/services/db_worker.py index fab4df2..7f58c01 100644 --- a/backend/app/services/db_worker.py +++ b/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 스레드를 막지 않고, - 백그라운드에서 큐에 쌓인 DB 작업만 순차적으로 처리하는 워커 - DB 커넥션 경합과 락(Lock)을 방지합니다. - """ - logger.info("🛠️ Background DB Writer Worker started") + 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": - # 매매일지 추가 - new_trade = TradeHistory(**data) - session.add(new_trade) + session.add(TradeHistory(**data)) 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 등에서 이관될 로직용 플레이스홀더) - # 실제 upsert 로직은 데이터 구조에 맞춰 구현 - logger.info(f"✅ [Worker] 종목분석 비동기 저장 처리 (Placeholder): {data.get('stock_code', 'Unknown')}") - + from app.services.analysis_db_service import _execute_upsert + await _execute_upsert(session, [data]) + 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: - logger.error(f"❌ [Worker] DB 작업 중 에러 발생: {e}", exc_info=True) + except Exception as exc: + logger.error("❌ [Worker] DB task failed: %s", exc, exc_info=True) + # 실패한 task 는 드롭 — 재시도가 필요하면 retry_count 추가 가능 finally: db_task_queue.task_done()