From 62fccd6d6ff44385f183c1034b62ac30f3b339ad Mon Sep 17 00:00:00 2001
From: YoungWooSeo <wepi98@gmail.com>
Date: Fri, 27 Feb 2026 12:28:07 +0900
Subject: [PATCH] =?UTF-8?q?fix:=20batch=5Fdaily=5Fanalysis=20=EB=A6=AC?=
 =?UTF-8?q?=EB=B7=B0=20=EC=9D=B4=EC=8A=88=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- [🔴] saver_task 예외 시 deadlock 방지: asyncio.wait로 saver_task 예외 감지
- [🔴] logging.basicConfig를 __main__ 블록으로 이동 (인프로세스 실행 시 충돌 방지)
- [🟡] get_all_tradable_stocks: 최소 종목 수(50개) 미만 시 RuntimeError 발생
- [🟠] detect_and_log_market_regime: df=None일 때 len(df) 호출 안전하게 수정
---
 backend/batch_daily_analysis.py | 114 +++++++++++++++++++++++---------
 1 file changed, 83 insertions(+), 31 deletions(-)

diff --git a/backend/batch_daily_analysis.py b/backend/batch_daily_analysis.py
index 5bee7b9..e9cf7e0 100644
--- a/backend/batch_daily_analysis.py
+++ b/backend/batch_daily_analysis.py
@@ -36,10 +36,8 @@ from app.services.analysis_db_service import bulk_upsert_analysis, passes_price_
 from app.api.endpoints.recommendations import analyze_stock_for_batch, StockRecommendation
 import logging
 
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-)
+# 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출
+#    인프로세스 스케줄러(main.py)로 실행될 때 로그 설정 충돌 방지
 logger = logging.getLogger(__name__)
 
 # -------------------------------------------------------------------
@@ -48,6 +46,9 @@ logger = logging.getLogger(__name__)
 BATCH_CONCURRENCY = 10    # API 세마포어 허용 범위 내 동시 처리 수
 BULK_SAVE_CHUNK   = 20    # Streaming 저장 단위 (큐에 20개 쌓이면 즉시 UPSERT)
 
+# 종목 목록 조회 실패 시 배치를 중단할 최소 종목 수
+MIN_STOCK_COUNT = 50
+
 
 # -------------------------------------------------------------------
 # 1. 시장 레짐 판별 (KOSPI 기준)
@@ -68,8 +69,11 @@ async def detect_and_log_market_regime() -> str:
 
         # KOSPI 지수 코드 1001
         df = pykrx_stock.get_index_ohlcv_by_date(start, today, "1001")
-        if df is None or len(df) < 205:
-            logger.warning("⚠️  KOSPI 데이터 부족(행 %d) — 레짐 판별 스킵", 0 if df is None else len(df))
+
+        # 🔧 [fix] df=None일 때 len(df) 직접 호출 방지
+        df_len = len(df) if df is not None else 0
+        if df_len < 205:
+            logger.warning("⚠️  KOSPI 데이터 부족(행 %d) — 레짐 판별 스킵", df_len)
             return "UNKNOWN"
 
         kospi_closes = df["종가"].tolist()
@@ -94,6 +98,9 @@ async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500)
 
     ETF / 레버리지 / 스팭 등 특수종목 자동 제외.
     KOSPI + KOSDAQ 합산 시가총액 상위 limit개 반환.
+
+    Raises:
+        RuntimeError: 조회 결과가 MIN_STOCK_COUNT(50) 미만인 경우 배치 중단
     """
     logger.info("Fetching top %d stocks by market cap from pykrx...", limit)
     try:
@@ -130,8 +137,18 @@ async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500)
         kospi_n  = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSPI")
         kosdaq_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSDAQ")
         logger.info("✅ 선정 %d개 (KOSPI %d / KOSDAQ %d)", len(top_stocks), kospi_n, kosdaq_n)
+
+        # 🔧 [fix] 최소 종목 수 검증 — 너무 적으면 잘못된 데이터로 배치 실행 방지
+        if len(top_stocks) < MIN_STOCK_COUNT:
+            raise RuntimeError(
+                f"종목 조회 결과 너무 적음: {len(top_stocks)}개 (최소 {MIN_STOCK_COUNT}개 필요). "
+                "pykrx API 오류 또는 장 휴장 여부를 확인하세요."
+            )
+
         return top_stocks
 
+    except RuntimeError:
+        raise  # 최소 종목 수 미달 — 상위로 전파하여 배치 중단
     except Exception as exc:
         logger.error("❌ 종목 목록 조회 실패: %s", exc, exc_info=True)
         return []
@@ -191,33 +208,44 @@ async def _streaming_saver(
     buffer: list[tuple[str, str, StockRecommendation]] = []
     stats = {"ok": 0, "no_chart": 0, "price_filter": 0, "error": 0, "saved": 0}
 
-    while True:
-        item = await queue.get()
-        if item is None:  # sentinel: 분석 완료
-            break
+    try:
+        while True:
+            item = await queue.get()
+            if item is None:  # sentinel: 분석 완료
+                break
 
-        status, code, name, result = item
-        stats[status] = stats.get(status, 0) + 1
+            status, code, name, result = item
+            stats[status] = stats.get(status, 0) + 1
 
-        if status == "ok" and result:
-            buffer.append((code, name, result))
+            if status == "ok" and result:
+                buffer.append((code, name, result))
 
-        # 버퍼가 청크 크기에 도달하면 즉시 저장
-        if len(buffer) >= BULK_SAVE_CHUNK:
+            # 버퍼가 청크 크기에 도달하면 즉시 저장
+            if len(buffer) >= BULK_SAVE_CHUNK:
+                async with AsyncSessionLocal() as session:
+                    saved = await bulk_upsert_analysis(session, buffer, analysis_date)
+                    await session.commit()
+                stats["saved"] += saved
+                logger.info("💾 DB 저장: %d건 누적", stats["saved"])
+                buffer.clear()
+
+        # 나머지 잔여분 저장
+        if buffer:
             async with AsyncSessionLocal() as session:
                 saved = await bulk_upsert_analysis(session, buffer, analysis_date)
                 await session.commit()
             stats["saved"] += saved
-            logger.info("💾 DB 저장: %d건 누적", stats["saved"])
-            buffer.clear()
+            logger.info("💾 DB 저장(잔여): %d건 누적", stats["saved"])
 
-    # 나머지 잔여분 저장
-    if buffer:
-        async with AsyncSessionLocal() as session:
-            saved = await bulk_upsert_analysis(session, buffer, analysis_date)
-            await session.commit()
-        stats["saved"] += saved
-        logger.info("💾 DB 저장(잔여): %d건 누적", stats["saved"])
+    except Exception as exc:
+        # 🔧 [fix] saver_task 예외 시 큐를 드레인하여 분석 태스크 deadlock 방지
+        logger.error("❌ _streaming_saver 예외 발생 — 큐 드레인 후 종료: %s", exc, exc_info=True)
+        # 큐에 남은 항목 모두 소비하여 analyze_one의 queue.put() 블로킹 해제
+        while not queue.empty():
+            try:
+                queue.get_nowait()
+            except asyncio.QueueEmpty:
+                break
 
     return stats
 
@@ -272,16 +300,34 @@ async def batch_analyze_all_stocks(limit: int = None):
             logger.error("❌ analyze_one unexpected error %s: %s", code, exc)
             await queue.put(("error", code, name, None))
 
-    # 저장기를 백그라운드로 먼저 시작
+    # 🔧 [fix] saver_task 예외 감지: asyncio.wait로 saver_task 비정상 종료 시 조기 감지
     saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date))
 
     # 분석 태스크 전체 실행
-    tasks = [analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)]
-    await asyncio.gather(*tasks, return_exceptions=True)
+    analysis_gather = asyncio.gather(
+        *[analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)],
+        return_exceptions=True,
+    )
 
-    # sentinel 전송 → 저장기 종료
-    await queue.put(None)
-    stats = await saver_task
+    # saver_task가 예외로 죽으면 analysis_gather도 조기 취소
+    done, pending = await asyncio.wait(
+        [asyncio.ensure_future(analysis_gather), saver_task],
+        return_when=asyncio.FIRST_EXCEPTION,
+    )
+
+    # saver_task 예외 발생 시 분석 태스크 취소 처리
+    for task in pending:
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, Exception):
+            pass
+
+    # sentinel 전송 → 저장기 종료 (정상 흐름에서도 항상 실행)
+    if not saver_task.done():
+        await queue.put(None)
+
+    stats = await saver_task if not saver_task.cancelled() else {}
 
     ok_count  = stats.get("ok", 0)
     no_chart  = stats.get("no_chart", 0)
@@ -344,4 +390,10 @@ async def main():
 
 
 if __name__ == "__main__":
+    # 🔧 [fix] basicConfig를 __main__ 블록으로 이동
+    #    인프로세스 스케줄러(main.py lifespan)에서 import될 때 로그 설정 충돌 방지
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
     asyncio.run(main())