""" 장중 미니배치: 매 10분마다 실행 (평일 09:00~15:30 KST) [흐름] 1. 아침 배치(batch_daily_analysis.py)가 분석종목(watchlist)을 score 상위 50개로 초기화 2. 장중 미니배치는 분석종목만 재분석하여 최신 지표로 stock_analysis 갱신 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 """ import asyncio import sys from datetime import datetime, date from pathlib import Path import pytz KST = pytz.timezone('Asia/Seoul') sys.path.insert(0, str(Path(__file__).parent)) from sqlalchemy import select, update, func from app.core.config import settings 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, bulk_upsert_analysis, passes_price_filter, run_db_scanner, ) # 🔧 단일화 from app.services.watchlist import ( WATCHLIST_MAX_SIZE, WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, ) import logging # 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출 logger = logging.getLogger(__name__) # 동시 처리 종목 수 (KIS API 세마포어가 추가로 속도 제어) 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]]: """현재 활성 분석종목 목록 조회.""" async with AsyncSessionLocal() as session: result = await session.execute( select(WatchlistStock.stock_code, WatchlistStock.stock_name) .where(WatchlistStock.is_active == True) ) rows = result.all() if not rows: logger.warning( "⚠️ 분석종목이 없습니다. " "아침 배치(batch_daily_analysis.py)가 먼저 실행되었는지 확인하세요." ) else: logger.info("📋 분석종목 %d개 대상", len(rows)) return [(row.stock_code, row.stock_name) for row in rows] # ------------------------------------------------------------------- # analyze_update_and_manage 분리: 분석 / 저장 / 관리 # ------------------------------------------------------------------- async def _save_analysis_and_history( session, stock_code: str, stock_name: str, result, analysis_date: date, now: datetime, ) -> None: """stock_analysis UPSERT + watchlist_history 시계열 기록. DELETE+INSERT race condition 없이 원자적으로 갱신. """ # 1. stock_analysis UPSERT (race-condition-free) await upsert_analysis(session, stock_code, stock_name, result, analysis_date) # 2. watchlist_history 시계열 기록 session.add(WatchlistHistory( stock_code=stock_code, stock_name=stock_name, recorded_at=now, score=result.score, current_price=result.current_price, change_rate=result.change_rate, rsi=result.rsi, signals=result.signals, )) async def _manage_watchlist_membership( session, stock_code: str, stock_name: str, score: int | None, now: datetime, ) -> bool: """score 기준 미달 시 watchlist에서 제거. Returns: True if removed, False if retained. """ if score is not None and score < WATCHLIST_EXIT_SCORE: await session.execute( update(WatchlistStock) .where( WatchlistStock.stock_code == stock_code, WatchlistStock.is_active == True, ) .values( is_active=False, removed_at=now, removed_reason="score_drop", ) ) logger.info( "🚫 %s (%s) 분석종목 제거: score=%s < %s", stock_code, stock_name, score, WATCHLIST_EXIT_SCORE, ) return True return False async def analyze_update_and_manage( stock_code: str, stock_name: str, analysis_date: date, ) -> tuple[bool, int | None, bool]: """종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → 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: # 🔧 [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) ) 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 # --- [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) return False, None, False # 5,000원 이하 저가주 → watchlist에서도 제거 if not passes_price_filter(result): logger.debug("Skip %s: price=%.0f원 < 5,000원", stock_code, result.current_price) async with AsyncSessionLocal() as session: await session.execute( update(WatchlistStock) .where( WatchlistStock.stock_code == stock_code, WatchlistStock.is_active == True, ) .values( is_active=False, removed_at=now, removed_reason="price_filter", ) ) await session.commit() return False, None, True # 가격필터 제거도 was_removed=True score = result.score async with AsyncSessionLocal() as session: await _save_analysis_and_history( session, stock_code, stock_name, result, analysis_date, now ) was_removed = await _manage_watchlist_membership( session, stock_code, stock_name, score, now ) await session.commit() logger.info("✅ %s (%s): score=%s", stock_code, stock_name, score) return True, score, was_removed except Exception as e: logger.error("❌ Failed %s: %s", stock_code, e) return False, None, False async def rescore_outside_candidates( analysis_date: date, current_codes: set[str], ) -> int: """워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신. [전략] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뛰. [최적화] 분석 결과를 모아서 bulk_upsert_analysis로 일괄 저장 (세션 최소화) 아침 배치에서 score 상위권이었던 종목이 장중 거래량·뉴스 급등으로 score가 올라가도 DB에는 오전 값이 그대로 남아 있다. 이를 포착하기 위해 워치리스트 밖 상위 OUTSIDE_RESCORE_LIMIT개를 재분석하여 stock_analysis를 최신화한다. 이후 refresh_watchlist()가 갱신된 score를 보고 신규 추가를 결정한다. Returns: 갱신 성공 종목 수 """ # 아침 배치 기준 score 상위 N개 중 워치리스트 미포함 종목 조회 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, StockAnalysis.score) .where( StockAnalysis.analysis_date == analysis_date, StockAnalysis.stock_code.notin_(exclude), ) .order_by(StockAnalysis.score.desc()) .limit(OUTSIDE_RESCORE_LIMIT) ) candidates = result.all() if not candidates: logger.info("외부 후보 종목 없음 — 재분석 건너뚁") return 0 total = len(candidates) logger.info("🔍 워치리스트 외부 후보 %d개 재분석 시작", total) semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) # 🔧 [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: logger.debug( " [%d/%d] ⏭️ AI 스킵 %s: score=%s < %s", i, total, code, db_score, MCP_ANALYSIS_MIN_SCORE, ) return None try: res = await analyze_stock_for_batch(code, name) if not res: return None if not passes_price_filter(res): logger.debug("Skip %s: price=%.0f원 < 5,000원", code, res.current_price) return None logger.debug(" [%d/%d] %s (%s): score=%s", i, total, code, name, res.score) return (code, name, res) except Exception as e: logger.error("❌ 외부 후보 재분석 실패 %s: %s", code, e) return None tasks = [ rescore_one(i, code, name, score) for i, (code, name, score) in enumerate(candidates, 1) ] 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 async def refresh_watchlist(analysis_date: date, current_codes: set[str]) -> int: """분석종목에 없는 고점수 종목을 신규 추가. Args: analysis_date: 오늘 날짜 (stock_analysis 기준) current_codes: 이번 배치 시작 시점의 watchlist 코드 집합 Returns: 추가된 종목 수 """ async with AsyncSessionLocal() as session: # 현재 활성 종목 수 확인 count_result = await session.execute( select(func.count()) .select_from(WatchlistStock) .where(WatchlistStock.is_active == True) ) active_count = count_result.scalar() or 0 slots = WATCHLIST_MAX_SIZE - active_count if slots <= 0: logger.debug("Watchlist 최대치 도달 (%d/%d)", active_count, WATCHLIST_MAX_SIZE) return 0 # stock_analysis에서 고점수 종목 중 현재 watchlist에 없는 것 조회 exclude_codes = list(current_codes) if current_codes else ["__none__"] result = await session.execute( select( StockAnalysis.stock_code, StockAnalysis.stock_name, StockAnalysis.score, ) .where( StockAnalysis.analysis_date == analysis_date, StockAnalysis.score >= WATCHLIST_ENTRY_SCORE, StockAnalysis.stock_code.notin_(exclude_codes), ) .order_by(StockAnalysis.score.desc()) .limit(slots) ) candidates = result.all() if not candidates: return 0 now = datetime.now(KST) for row in candidates: session.add(WatchlistStock( stock_code=row.stock_code, stock_name=row.stock_name, added_at=now, added_reason="score_rise", score_at_add=row.score, is_active=True, )) await session.commit() logger.info( "➕ 분석종목 %d개 신규 추가 (score ≥ %d, 현재 %d개)", len(candidates), WATCHLIST_ENTRY_SCORE, active_count + len(candidates), ) return len(candidates) async def main(): """장중 미니배치 메인""" # 장중(09:00~15:30 KST 평일)이 아니면 즉시 종료 if not is_market_open(): logger.info("📴 장중이 아닙니다 (09:00~15:30 평일). 종료합니다.") return start_time = datetime.now(KST) analysis_date = datetime.now(KST).date() logger.info( "⚡ [Intraday] %s 시작 — 분석종목 재분석 (진입≥%d, 제거<%d, AI호출≥%d점)", start_time.strftime("%H:%M:%S"), WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, MCP_ANALYSIS_MIN_SCORE, ) # 활성 분석종목 조회 active_stocks = await get_active_watchlist() if not active_stocks: return # 이번 배치 시작 시점의 코드 집합 (신규 추가 시 중복 방지용) current_codes = {code for code, _ in active_stocks} total = len(active_stocks) # 병렬 재분석 semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) async def analyze_one(i: int, code: str, name: str) -> tuple[bool, int | None, bool]: async with semaphore: logger.info("[%d/%d] %s (%s)", i, total, code, name) return await analyze_update_and_manage(code, name, analysis_date) tasks = [ analyze_one(i, code, name) for i, (code, name) in enumerate(active_stocks, 1) ] 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) # 🔧 [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 ) # 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) # 장중 DB 스캐너 실행 — 장중 과매도 신규 종목 포착 (RSI≤30 + BB하단 + ADX≤30 + MA방향) scanner_tickers = await run_db_scanner(analysis_date) scanner_new = scanner_tickers - current_codes # 워치리스트 미포함 신규 신호 종목 added_scanner = 0 if scanner_new: now_kst = datetime.now(KST) async with AsyncSessionLocal() as session: # 현재 활성 종목 수 확인 count_result = await session.execute( select(func.count()) .select_from(WatchlistStock) .where(WatchlistStock.is_active == True) ) active_count = count_result.scalar() or 0 slots = WATCHLIST_MAX_SIZE - active_count for code in list(scanner_new)[:slots]: # stock_analysis에서 종목명 조회 name_result = await session.execute( select(StockAnalysis.stock_name) .where( StockAnalysis.stock_code == code, StockAnalysis.analysis_date == analysis_date, ) .limit(1) ) name_row = name_result.first() if not name_row: continue session.add(WatchlistStock( stock_code=code, stock_name=name_row.stock_name, added_at=now_kst, added_reason="intraday_scanner", score_at_add=None, is_active=True, )) added_scanner += 1 await session.commit() if added_scanner: logger.info("📡 장중 스캐너 신규 추가: %d개 (RSI≤30+BB하단)", added_scanner) current_codes |= scanner_new # refresh_watchlist 중복 방지용 갱신 # 고점수 신규 종목 추가 (스캐너 추가분 제외) added = await refresh_watchlist(analysis_date, current_codes) duration = (datetime.now(KST) - start_time).total_seconds() logger.info( "✅ [Intraday] 완료: %d/%d 유지, %d 제거, +%d 신규(score), +%d 신규(스캐너), " "%d AI스킵(score<%d), %d 실패, %d 오류 — %.1f초", retained, total, removed, added, added_scanner, ai_skipped, MCP_ANALYSIS_MIN_SCORE, fail, error, duration, ) if __name__ == "__main__": # 🔧 [fix] basicConfig를 __main__ 블록으로 이동 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) asyncio.run(main())