From de8645b3bf3b6a981b3f229f5de9168cee52997a Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Thu, 26 Feb 2026 11:41:57 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20MCP=20AI=20=EB=B6=84=EC=84=9D=EC=9D=84?= =?UTF-8?q?=20score=20>=3D=20MCP=5FANALYSIS=5FMIN=5FSCORE(70)=20=EC=A2=85?= =?UTF-8?q?=EB=AA=A9=EB=A7=8C=20=ED=98=B8=EC=B6=9C=ED=95=98=EB=8F=84?= =?UTF-8?q?=EB=A1=9D=20=EC=B5=9C=EC=A0=81=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/batch_intraday_analysis.py | 82 ++++++++++++++++++++++++++---- 1 file changed, 73 insertions(+), 9 deletions(-) diff --git a/backend/batch_intraday_analysis.py b/backend/batch_intraday_analysis.py index 54724f5..6a62427 100644 --- a/backend/batch_intraday_analysis.py +++ b/backend/batch_intraday_analysis.py @@ -7,6 +7,10 @@ 3. 분석 결과를 watchlist_history에 시계열로 누적 저장 4. score 기준으로 분석종목 진입(≥60) / 제거(<20) 자동 관리 +[최적화] +- MCP AI 분석은 DB 점수가 MCP_ANALYSIS_MIN_SCORE(기본 70) 이상인 종목만 호출 +- 저점수 종목은 루트림 지표 계산만 수행하고 수동 score 유지 + 실행: python batch_intraday_analysis.py Cron: */10 9-15 * * 1-5 docker exec autostock-backend python /app/batch_intraday_analysis.py >> /logs/batch_intraday.log 2>&1 """ @@ -48,6 +52,9 @@ BATCH_CONCURRENCY = 5 # 워치리스트 외부 후보 재분석 상한 (10분 배치 내에 처리 가능한 수준) OUTSIDE_RESCORE_LIMIT = 30 +# MCP AI 분석 호출 최소 점수 (settings에서 읽음) +MCP_ANALYSIS_MIN_SCORE: int = settings.MCP_ANALYSIS_MIN_SCORE + async def get_active_watchlist() -> list[tuple[str, str]]: """현재 활성 분석종목 목록 조회.""" @@ -68,6 +75,24 @@ 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 분리: 분석 / 저장 / 관리 # ------------------------------------------------------------------- @@ -139,10 +164,33 @@ 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 적스 체크만 수행 + - 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, + ) + # watchlist 제거 여부만 체크 (WATCHLIST_EXIT_SCORE 미만이면 제거) + now = datetime.now(KST) + async with AsyncSessionLocal() as session: + was_removed = await _manage_watchlist_membership( + session, stock_code, stock_name, current_score, now + ) + await session.commit() + return True, current_score, was_removed + + # --- [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) @@ -193,6 +241,8 @@ async def rescore_outside_candidates( ) -> int: """워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신. + [최적화] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뜀. + 아침 배치에서 score 상위권이었던 종목이 장중 거래량·뉴스 급등으로 score가 올라가도 DB에는 오전 값이 그대로 남아 있다. 이를 포착하기 위해 워치리스트 밖 상위 OUTSIDE_RESCORE_LIMIT개를 @@ -205,7 +255,7 @@ async def rescore_outside_candidates( async with AsyncSessionLocal() as session: exclude = list(current_codes) if current_codes else ["__none__"] result = await session.execute( - select(StockAnalysis.stock_code, StockAnalysis.stock_name) + select(StockAnalysis.stock_code, StockAnalysis.stock_name, StockAnalysis.score) .where( StockAnalysis.analysis_date == analysis_date, StockAnalysis.stock_code.notin_(exclude), @@ -224,8 +274,15 @@ async def rescore_outside_candidates( semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) - async def rescore_one(i: int, code: str, name: str) -> bool: + async def rescore_one(i: int, code: str, name: str, db_score: int | None) -> bool: async with semaphore: + # DB score < MCP_ANALYSIS_MIN_SCORE 이면 AI 호출 스킵 + if db_score is not None and db_score < MCP_ANALYSIS_MIN_SCORE: + logger.debug( + " [%d/%d] ⏭️ AI 스킵 %s: score=%s < %s", + i, total, code, db_score, MCP_ANALYSIS_MIN_SCORE, + ) + return False try: res = await analyze_stock_for_batch(code, name) if not res: @@ -246,12 +303,13 @@ async def rescore_outside_candidates( return False tasks = [ - rescore_one(i, code, name) - for i, (code, name) in enumerate(candidates, 1) + 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) - logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신", success, total) + skipped = sum(1 for r in results if r is False) + logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신, %d개 AI 스킵", success, total, skipped) return success @@ -328,8 +386,8 @@ async def main(): start_time = datetime.now(KST) analysis_date = datetime.now(KST).date() logger.info( - "⚡ [Intraday] %s 시작 — 분석종목 재분석 (진입≥%d, 제거<%d)", - start_time.strftime("%H:%M:%S"), WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, + "⚡ [Intraday] %s 시작 — 분석종목 재분석 (진입≥%d, 제거<%d, AI호출≥%d점)", + start_time.strftime("%H:%M:%S"), WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, MCP_ANALYSIS_MIN_SCORE, ) # 활성 분석종목 조회 @@ -362,6 +420,11 @@ async def main(): ) 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_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 + ) # 워치리스트 외부 후보 재분석 — score DB 최신화 (refresh_watchlist 전에 실행) await rescore_outside_candidates(analysis_date, current_codes) @@ -416,8 +479,9 @@ async def main(): duration = (datetime.now(KST) - start_time).total_seconds() logger.info( "✅ [Intraday] 완료: %d/%d 유지, %d 제거, +%d 신규(score), +%d 신규(스캐너), " - "%d 실패, %d 오류 — %.1f초", - retained, total, removed, added, added_scanner, fail, error, duration, + "%d AI스킵(score<%d), %d 실패, %d 오류 — %.1f초", + retained, total, removed, added, added_scanner, + ai_skipped, MCP_ANALYSIS_MIN_SCORE, fail, error, duration, )