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())