From a47591e40d7836d41d333d6746354ce79853ef73 Mon Sep 17 00:00:00 2001
From: YoungWooSeo <wepi98@gmail.com>
Date: Fri, 27 Feb 2026 12:30:56 +0900
Subject: [PATCH] =?UTF-8?q?fix:=20batch=5Fintraday=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

- [🔴] logging.basicConfig를 __main__ 블록으로 이동
- [🟡] analyze_update_and_manage: get_current_score 세션 통합 (세션 인스턴스 감소)
- [🟡] rescore_outside_candidates: bulk_upsert_analysis로 일괄 저장
- [🟠] retained 통계: AI 스킵 종목 명확히 분리
---
 backend/batch_intraday_analysis.py | 149 ++++++++++++++++-------------
 1 file changed, 85 insertions(+), 64 deletions(-)

diff --git a/backend/batch_intraday_analysis.py b/backend/batch_intraday_analysis.py
index 6a62427..b95c38d 100644
--- a/backend/batch_intraday_analysis.py
+++ b/backend/batch_intraday_analysis.py
@@ -7,7 +7,7 @@
 3. 분석 결과를 watchlist_history에 시계열로 누적 저장
 4. score 기준으로 분석종목 진입(≥60) / 제거(<20) 자동 관리
 
-[최적화]
+[전략]
 - MCP AI 분석은 DB 점수가 MCP_ANALYSIS_MIN_SCORE(기본 70) 이상인 종목만 호출
 - 저점수 종목은 루트림 지표 계산만 수행하고 수동 score 유지
 
@@ -32,7 +32,12 @@ from app.core.database import AsyncSessionLocal          # 🔧 공통 DB 모듈
 from app.core.market_schedule import is_market_open
 from app.models.models import StockAnalysis, WatchlistStock, WatchlistHistory
 from app.api.endpoints.recommendations import analyze_stock_for_batch
-from app.services.analysis_db_service import upsert_analysis, passes_price_filter, run_db_scanner  # 🔧 단일화
+from app.services.analysis_db_service import (
+    upsert_analysis,
+    bulk_upsert_analysis,
+    passes_price_filter,
+    run_db_scanner,
+)  # 🔧 단일화
 from app.services.watchlist import (
     WATCHLIST_MAX_SIZE,
     WATCHLIST_ENTRY_SCORE,
@@ -40,10 +45,7 @@ from app.services.watchlist import (
 )
 import logging
 
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-)
+# 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출
 logger = logging.getLogger(__name__)
 
 # 동시 처리 종목 수 (KIS API 세마포어가 추가로 속도 제어)
@@ -75,24 +77,6 @@ async def get_active_watchlist() -> list[tuple[str, str]]:
     return [(row.stock_code, row.stock_name) for row in rows]
 
 
-async def get_current_score(stock_code: str, analysis_date: date) -> int | None:
-    """최신 DB score 조회.
-
-    MCP AI 분석 전 score를 확인하여 MCP_ANALYSIS_MIN_SCORE 미만은 AI 호출을 스킵하기 위해 사용.
-    """
-    async with AsyncSessionLocal() as session:
-        result = await session.execute(
-            select(StockAnalysis.score)
-            .where(
-                StockAnalysis.stock_code == stock_code,
-                StockAnalysis.analysis_date == analysis_date,
-            )
-            .limit(1)
-        )
-        row = result.first()
-    return row.score if row else None
-
-
 # -------------------------------------------------------------------
 # analyze_update_and_manage 분리: 분석 / 저장 / 관리
 # -------------------------------------------------------------------
@@ -164,33 +148,47 @@ async def analyze_update_and_manage(
 ) -> tuple[bool, int | None, bool]:
     """종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → watchlist 관리.
 
-    [최적화] DB score < MCP_ANALYSIS_MIN_SCORE이면 AI 분석 스킵.
-      - AI 호출 없이 기존 score 유지, watchlist 적스 체크만 수행
+    [전략] DB score < MCP_ANALYSIS_MIN_SCORE이면 AI 분석 스킵.
+      - AI 호출 없이 기존 score 유지, watchlist 진입 체크만 수행
       - score 개선 시코어는 다음 주기에 반영됨
 
+    [포인트] 세션을 한 번만 열어 score 조회 + 저장 + 관리를 한 트랜잭션에서 처리
+
     Returns: (success, score, was_removed)
       was_removed: _manage_watchlist_membership이 실제로 비활성화 처리했으면 True
     """
     try:
-        # --- [1단계] DB에서 현재 score 조회 ---
-        current_score = await get_current_score(stock_code, analysis_date)
-
-        # --- [2단계] score 미달 시 AI 분석 스킵 ---
-        if current_score is not None and current_score < MCP_ANALYSIS_MIN_SCORE:
-            logger.debug(
-                "⏭️  AI 스킵 %s (%s): score=%s < %s",
-                stock_code, stock_name, current_score, MCP_ANALYSIS_MIN_SCORE,
+        # 🔧 [fix] score 조회 + 저장 + watchlist 관리를 단일 세션으로 통합
+        #    기존에는 get_current_score에서 세션 1번 + 저장에서 세션 1번으로 최소 2번 열렸으나
+        #    이제 score 조회부터 저장까지 단일 세션에서 모두 처리
+        now = datetime.now(KST)
+        async with AsyncSessionLocal() as session:
+            # --- [1단계] DB에서 현재 score 조회 ---
+            score_result = await session.execute(
+                select(StockAnalysis.score)
+                .where(
+                    StockAnalysis.stock_code == stock_code,
+                    StockAnalysis.analysis_date == analysis_date,
+                )
+                .limit(1)
             )
-            # watchlist 제거 여부만 체크 (WATCHLIST_EXIT_SCORE 미만이면 제거)
-            now = datetime.now(KST)
-            async with AsyncSessionLocal() as session:
+            score_row = score_result.first()
+            current_score = score_row.score if score_row else None
+
+            # --- [2단계] score 미달 시 AI 분석 스킵 ---
+            if current_score is not None and current_score < MCP_ANALYSIS_MIN_SCORE:
+                logger.debug(
+                    "⏭️  AI 스킵 %s (%s): score=%s < %s",
+                    stock_code, stock_name, current_score, MCP_ANALYSIS_MIN_SCORE,
+                )
+                # watchlist 제거 여부만 체크 (WATCHLIST_EXIT_SCORE 미만이면 제거)
                 was_removed = await _manage_watchlist_membership(
                     session, stock_code, stock_name, current_score, now
                 )
                 await session.commit()
-            return True, current_score, was_removed
+                return True, current_score, was_removed
 
-        # --- [3단계] score 기준 이상 → AI 분석 실행 ---
+        # --- [3단계] score 기준 이상 → AI 분석 실행 (세션 밖에서 실행) ---
         result = await analyze_stock_for_batch(stock_code, stock_name)
         if not result:
             logger.info("No result for %s (chart/price issue)", stock_code)
@@ -208,14 +206,13 @@ async def analyze_update_and_manage(
                     )
                     .values(
                         is_active=False,
-                        removed_at=datetime.now(KST),
+                        removed_at=now,
                         removed_reason="price_filter",
                     )
                 )
                 await session.commit()
             return False, None, True  # 가격필터 제거도 was_removed=True
 
-        now = datetime.now(KST)
         score = result.score
 
         async with AsyncSessionLocal() as session:
@@ -241,7 +238,8 @@ async def rescore_outside_candidates(
 ) -> int:
     """워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신.
 
-    [최적화] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뜀.
+    [전략] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뛰.
+    [최적화] 분석 결과를 모아서 bulk_upsert_analysis로 일괄 저장 (세션 최소화)
 
     아침 배치에서 score 상위권이었던 종목이 장중 거래량·뉴스 급등으로
     score가 올라가도 DB에는 오전 값이 그대로 남아 있다.
@@ -266,7 +264,7 @@ async def rescore_outside_candidates(
         candidates = result.all()
 
     if not candidates:
-        logger.info("외부 후보 종목 없음 — 재분석 건너뜀")
+        logger.info("외부 후보 종목 없음 — 재분석 건너뚁")
         return 0
 
     total = len(candidates)
@@ -274,7 +272,14 @@ async def rescore_outside_candidates(
 
     semaphore = asyncio.Semaphore(BATCH_CONCURRENCY)
 
-    async def rescore_one(i: int, code: str, name: str, db_score: int | None) -> bool:
+    # 🔧 [fix] 분석 결과를 수집하여 bulk_upsert_analysis로 일괄 저장
+    #    기존: rescore_one마다 세션 1개 open/close (30개 대상 시 30번)
+    #    개선: 모아서 한 번에 bulk upsert (1번)
+
+    async def rescore_one(
+        i: int, code: str, name: str, db_score: int | None
+    ) -> tuple[str, str, object] | None:
+        """Returns (code, name, result) on success, None otherwise."""
         async with semaphore:
             # DB score < MCP_ANALYSIS_MIN_SCORE 이면 AI 호출 스킵
             if db_score is not None and db_score < MCP_ANALYSIS_MIN_SCORE:
@@ -282,33 +287,39 @@ async def rescore_outside_candidates(
                     "  [%d/%d] ⏭️ AI 스킵 %s: score=%s < %s",
                     i, total, code, db_score, MCP_ANALYSIS_MIN_SCORE,
                 )
-                return False
+                return None
             try:
                 res = await analyze_stock_for_batch(code, name)
                 if not res:
-                    return False
-                # 저가주 필터 (CLAUDE.md 5절 가격 필터)
+                    return None
                 if not passes_price_filter(res):
                     logger.debug("Skip %s: price=%.0f원 < 5,000원", code, res.current_price)
-                    return False
-
-                async with AsyncSessionLocal() as session:
-                    await upsert_analysis(session, code, name, res, analysis_date)
-                    await session.commit()
-
+                    return None
                 logger.debug("  [%d/%d] %s (%s): score=%s", i, total, code, name, res.score)
-                return True
+                return (code, name, res)
             except Exception as e:
                 logger.error("❌ 외부 후보 재분석 실패 %s: %s", code, e)
-                return False
+                return None
 
     tasks = [
         rescore_one(i, code, name, score)
         for i, (code, name, score) in enumerate(candidates, 1)
     ]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-    success = sum(1 for r in results if r is True)
-    skipped = sum(1 for r in results if r is False)
+    raw_results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    # 유효한 결과만 필터막한 후 일괄 저장
+    valid_items = [
+        r for r in raw_results
+        if isinstance(r, tuple) and len(r) == 3
+    ]
+    success = len(valid_items)
+    skipped = len(candidates) - success
+
+    if valid_items:
+        async with AsyncSessionLocal() as session:
+            await bulk_upsert_analysis(session, valid_items, analysis_date)
+            await session.commit()
+
     logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신, %d개 AI 스킵", success, total, skipped)
     return success
 
@@ -413,18 +424,23 @@ async def main():
     ]
     results = await asyncio.gather(*tasks, return_exceptions=True)
 
-    analyzed_ok  = sum(1 for r in results if isinstance(r, tuple) and r[0] is True)
-    removed = sum(1 for r in results if isinstance(r, tuple) and len(r) == 3 and r[2] is True)
-    retained = analyzed_ok - sum(
-        1 for r in results if isinstance(r, tuple) and len(r) == 3 and r[0] is True and r[2] is True
+    analyzed_ok = sum(1 for r in results if isinstance(r, tuple) and r[0] is True)
+    removed     = sum(1 for r in results if isinstance(r, tuple) and len(r) == 3 and r[2] is True)
+
+    # 🔧 [fix] retained: AI스킵이지만 성공한 종목은 유지(제거 X)로 계산
+    #    was_removed=True인 네 가지 코드(success=True, removed=True)를 제외만 한 것
+    retained = sum(
+        1 for r in results
+        if isinstance(r, tuple) and len(r) == 3 and r[0] is True and r[2] is False
     )
-    fail  = sum(1 for r in results if isinstance(r, tuple) and r[0] is False)
-    error = sum(1 for r in results if isinstance(r, Exception))
+    # AI 스킵 종목 수: success=True이고 score < MCP_ANALYSIS_MIN_SCORE인 것
     ai_skipped = sum(
         1 for r in results
         if isinstance(r, tuple) and r[0] is True
         and r[1] is not None and r[1] < MCP_ANALYSIS_MIN_SCORE
     )
+    fail  = sum(1 for r in results if isinstance(r, tuple) and r[0] is False)
+    error = sum(1 for r in results if isinstance(r, Exception))
 
     # 워치리스트 외부 후보 재분석 — score DB 최신화 (refresh_watchlist 전에 실행)
     await rescore_outside_candidates(analysis_date, current_codes)
@@ -486,4 +502,9 @@ async def main():
 
 
 if __name__ == "__main__":
+    # 🔧 [fix] basicConfig를 __main__ 블록으로 이동
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
     asyncio.run(main())