diff --git a/backend/batch_intraday_analysis.py b/backend/batch_intraday_analysis.py index 07e28c5..1747bd4 100644 --- a/backend/batch_intraday_analysis.py +++ b/backend/batch_intraday_analysis.py @@ -42,6 +42,9 @@ 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://"): @@ -192,6 +195,103 @@ async def analyze_update_and_manage( 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: """분석종목에 없는 고점수 종목을 신규 추가. @@ -301,6 +401,9 @@ 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)) + # 워치리스트 외부 후보 재분석 — score DB 최신화 (refresh_watchlist 전에 실행) + await rescore_outside_candidates(analysis_date, current_codes) + # 고점수 신규 종목 추가 added = await refresh_watchlist(analysis_date, current_codes)