""" 장중 미니배치: 매 10분마다 실행 (평일 09:00~15:30 KST) [흐름] 1. 아침 배치(batch_daily_analysis.py)가 분석종목(watchlist)을 score 상위 50개로 초기화 2. 장중 미니배치는 분석종목만 재분석하여 최신 지표로 stock_analysis 갱신 3. 분석 결과를 watchlist_history에 시계열로 누적 저장 4. score 기준으로 분석종목 진입(≥60) / 제거(<20) 자동 관리 실행: 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 sys.path.insert(0, str(Path(__file__).parent)) from sqlalchemy import select, delete, update, func from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker import logging from app.core.config import settings 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.watchlist import ( WATCHLIST_MAX_SIZE, WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, ) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # 동시 처리 종목 수 (KIS API 세마포어가 추가로 속도 제어) BATCH_CONCURRENCY = 5 # 워치리스트 외부 후보 재분석 상한 (10분 배치 내에 처리 가능한 수준) OUTSIDE_RESCORE_LIMIT = 30 # DB 연결 database_url = settings.DATABASE_URL if database_url.startswith("postgresql://"): database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1) engine = create_async_engine( database_url, echo=False, pool_pre_ping=True, pool_size=5, max_overflow=10, ) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False, ) 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(f"📋 분석종목 {len(rows)}개 대상") return [(row.stock_code, row.stock_name) for row in rows] async def analyze_update_and_manage( stock_code: str, stock_name: str, analysis_date: date, ) -> tuple[bool, int | None]: """종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → watchlist 관리. Returns: (success, score) """ try: result = await analyze_stock_for_batch(stock_code, stock_name) if not result: logger.info(f"No result for {stock_code} (chart/price issue)") return False, None # 5,000원 이하 저가주 → watchlist에서도 제거 if result.current_price and result.current_price < 5000: logger.debug(f"Skip {stock_code}: price={result.current_price:.0f}원 < 5,000원") 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=datetime.now(), removed_reason="price_filter", ) ) await session.commit() return False, None now = datetime.now() score = result.score async with AsyncSessionLocal() as session: # 1. stock_analysis 갱신 (DELETE + INSERT) await session.execute( delete(StockAnalysis).where( StockAnalysis.stock_code == stock_code, StockAnalysis.analysis_date == analysis_date, ) ) session.add(StockAnalysis( stock_code=stock_code, stock_name=stock_name, analysis_date=analysis_date, current_price=result.current_price, change_rate=result.change_rate, volume=result.volume, rsi=result.rsi, ma5=result.ma5, ma20=result.ma20, ma60=result.ma60, stochastic_k=result.stochastic_k, stochastic_d=result.stochastic_d, bollinger_upper=result.bollinger_upper, bollinger_lower=result.bollinger_lower, bollinger_middle=result.bollinger_middle, signals=result.signals, score=score, ml_score=result.ml_score, ml_confidence=result.ml_confidence, ml_signal=result.ml_signal, atr_14=result.atr_14, adx_14=result.adx_14, )) # 2. watchlist_history 시계열 기록 session.add(WatchlistHistory( stock_code=stock_code, stock_name=stock_name, recorded_at=now, score=score, current_price=result.current_price, change_rate=result.change_rate, rsi=result.rsi, signals=result.signals, )) # 3. score가 기준 이하면 분석종목에서 제거 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( f"🚫 {stock_code} ({stock_name}) 분석종목 제거: " f"score={score} < {WATCHLIST_EXIT_SCORE}" ) await session.commit() logger.info(f"✅ {stock_code} ({stock_name}): score={score}") return True, score except Exception as e: logger.error(f"❌ Failed {stock_code}: {e}") return False, None async def rescore_outside_candidates( analysis_date: date, current_codes: set[str], ) -> int: """워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신. 아침 배치에서 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) .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(f"🔍 워치리스트 외부 후보 {total}개 재분석 시작") semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) async def rescore_one(i: int, code: str, name: str) -> bool: async with semaphore: try: res = await analyze_stock_for_batch(code, name) if not res: return False # 저가주 필터 (CLAUDE.md 섹션 6.4) if res.current_price and res.current_price < 5000: logger.debug(f"Skip {code}: price={res.current_price:.0f}원 < 5,000원") return False async with AsyncSessionLocal() as session: await session.execute( delete(StockAnalysis).where( StockAnalysis.stock_code == code, StockAnalysis.analysis_date == analysis_date, ) ) session.add(StockAnalysis( stock_code=code, stock_name=name, analysis_date=analysis_date, current_price=res.current_price, change_rate=res.change_rate, volume=res.volume, rsi=res.rsi, ma5=res.ma5, ma20=res.ma20, ma60=res.ma60, stochastic_k=res.stochastic_k, stochastic_d=res.stochastic_d, bollinger_upper=res.bollinger_upper, bollinger_lower=res.bollinger_lower, bollinger_middle=res.bollinger_middle, signals=res.signals, score=res.score, ml_score=res.ml_score, ml_confidence=res.ml_confidence, ml_signal=res.ml_signal, atr_14=res.atr_14, adx_14=res.adx_14, )) await session.commit() logger.debug(f" [{i}/{total}] {code} ({name}): score={res.score}") return True except Exception as e: logger.error(f"❌ 외부 후보 재분석 실패 {code}: {e}") return False tasks = [ rescore_one(i, code, name) for i, (code, name) 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(f"✅ 외부 후보 재분석 완료: {success}/{total}개 갱신") 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(f"Watchlist 최대치 도달 ({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() 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( f"➕ 분석종목 {len(candidates)}개 신규 추가 " f"(score ≥ {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() analysis_date = date.today() logger.info( f"⚡ [Intraday] {start_time.strftime('%H:%M:%S')} 시작 — " f"분석종목 재분석 (진입≥{WATCHLIST_ENTRY_SCORE}, 제거<{WATCHLIST_EXIT_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]: async with semaphore: logger.info(f"[{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) success = 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 r[0] is True and r[1] is not None and r[1] < WATCHLIST_EXIT_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) # 고점수 신규 종목 추가 added = await refresh_watchlist(analysis_date, current_codes) duration = (datetime.now() - start_time).total_seconds() logger.info( f"✅ [Intraday] 완료: {success}/{total} 성공, " f"{removed} 제거, +{added} 신규, " f"{fail} 실패, {error} 오류 — {duration:.1f}초" ) if __name__ == "__main__": asyncio.run(main())