""" 매일 아침 장 시작 전(9시) 실행되는 배치 분석 스크립트 전체 거래 종목을 분석하여 DB에 저장 1. 시장 레짐 판별 (KOSPI200 기준 BULL/BEAR/SIDEWAYS) 2. 차트 데이터 조회 + 기술적 지표 계산 3. 매매 신호 생성 + DB Streaming 저장 (분석 즉시 UPSERT) 4. 배치 완료 후 DB 스캐너 실행 (추가 API 호출 없이) 5. 워치리스트 초기화 — 스캐너 신호 종목 우선 배치 성능 개선 (이전 대비): - BATCH_CONCURRENCY: 5 → 10 (API 세마포어 허용 범위 내) - DB 트랜잭션: 종목당 2개(DELETE+INSERT) → UPSERT (원자적, race-condition 없음) - Streaming: 전체 메모리 적재 → 분석 완료 즉시 BULK_SAVE_CHUNK(20)개 단위 저장 - DB 엔진: 스크립트 자체 생성 → app.core.database 공통 모듈 사용 실행: python batch_daily_analysis.py Cron: 0 9 * * 1-5 cd /path/to/auto-stock/backend && python batch_daily_analysis.py """ import asyncio import sys from datetime import datetime, date from pathlib import Path from typing import Optional import pytz KST = pytz.timezone('Asia/Seoul') sys.path.insert(0, str(Path(__file__).parent)) from app.core.config import settings from app.core.database import AsyncSessionLocal # 🔧 공통 DB 모듈 사용 from app.services.korea_investment_api import KoreaInvestmentAPI, korea_investment_api from app.services.analysis_db_service import bulk_upsert_analysis, passes_price_filter, run_db_scanner # 🔧 단일화 from app.api.endpoints.recommendations import analyze_stock_for_batch, StockRecommendation import logging # 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출 # 인프로세스 스케줄러(main.py)로 실행될 때 로그 설정 충돌 방지 logger = logging.getLogger(__name__) # ------------------------------------------------------------------- # 성능 설정 # ------------------------------------------------------------------- BATCH_CONCURRENCY = 10 # API 세마포어 허용 범위 내 동시 처리 수 BULK_SAVE_CHUNK = 20 # Streaming 저장 단위 (큐에 20개 쌓이면 즉시 UPSERT) # 종목 목록 조회 실패 시 배치를 중단할 최소 종목 수 MIN_STOCK_COUNT = 50 # ------------------------------------------------------------------- # 1. 시장 레짐 판별 (KOSPI 기준) # ------------------------------------------------------------------- async def detect_and_log_market_regime() -> str: """배치 시작 시 KOSPI 200일 데이터로 시장 레짐 판별. BULL / BEAR / SIDEWAYS 중 하나를 리턴하며 배치 전체에 로깅된다. 실패 시 UNKNOWN을 리턴하여 배치를 차단하지 않는다. pykrx 대신 KIS API 사용 — KODEX 200(069500)으로 KOSPI 레짐 판별. KODEX 200은 KOSPI 200을 추종하므로 지수 대용으로 적합하다. """ try: from app.services.mean_reversion_strategy import detect_market_regime # KODEX 200(069500) 일봉 240개 = 약 1년치 데이터 data = await korea_investment_api.get_daily_chart("069500", period="D", limit=240) output = data.get("output2") or [] if not isinstance(output, list) or len(output) < 205: logger.warning("⚠️ KODEX200 데이터 부족(%d행) — 레짐 판별 스킵", len(output) if output else 0) return "UNKNOWN" # KIS API는 최신순 반환 → 오래된 순(시계열 순)으로 역정렬 closes = [float(item["stck_clpr"]) for item in reversed(output) if item.get("stck_clpr")] regime = detect_market_regime(closes) regime_emoji = {"상승장": "📈 BULL", "하락장": "📉 BEAR", "bull": "📈 BULL", "bear": "📉 BEAR"}.get(regime.value, "➡️ " + regime.value.upper()) logger.info("🌏 시장 레짐: %s (KODEX200 %d일 기준)", regime_emoji, len(closes)) return regime.value.upper() except Exception as exc: logger.error("❌ 레짐 판별 실패: %s", exc, exc_info=True) return "UNKNOWN" # ------------------------------------------------------------------- # 2. 종목 목록 조회 # ------------------------------------------------------------------- async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500) -> list[tuple[str, str]]: """거래량 상위 종목 조회 (KIS API 사용) pykrx 의존성 제거 — KIS 거래량 순위 API(FHPST01710000) 사용. 시가총액 기준 → 거래량 기준으로 변경 (유동성 높은 종목 우선). ETF / 레버리지 / 스팩 등 특수종목 자동 제외. KOSPI + KOSDAQ 합산 거래량 상위 limit개 반환. Raises: RuntimeError: 조회 결과가 MIN_STOCK_COUNT(50) 미만인 경우 배치 중단 """ logger.info("Fetching top %d stocks by volume from KIS API...", limit) SKIP_KEYWORDS = [ "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG", "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "선물", "스팩", "리츠", "SPAC", ] # 모의투자 모드에서 실계좌 시세 API를 사용할 수 있으면 준비 real_market_api: Optional[KoreaInvestmentAPI] = None if settings.KIS_MODE == "virtual" and settings.KIS_REAL_APP_KEY: try: real_market_api = KoreaInvestmentAPI.for_real_market_data() logger.info("📡 KOSDAQ 조회: 실계좌 시세 API 사용 (모의투자 서버 제한 우회)") except Exception as exc: logger.warning("⚠️ 실계좌 시세 API 초기화 실패, 모의투자 API 사용: %s", exc) stocks_with_vol = [] for market_code, market_name in [("J", "KOSPI"), ("Q", "KOSDAQ")]: # KOSDAQ은 실계좌 시세 API 우선 사용 (모의투자 서버는 KOSDAQ 미지원) api_to_use = real_market_api if (market_code == "Q" and real_market_api) else kis_api try: result = await api_to_use.get_volume_rank(market_code) items = result.get("output") or [] if not isinstance(items, list): logger.warning(" %s: output이 리스트 아님 — 건너뜀", market_name) continue count = 0 for item in items: ticker = item.get("mksc_shrn_iscd") or item.get("stck_shrn_iscd") or "" name = item.get("hts_kor_isnm") or item.get("stck_kor_isnm") or "" volume = int(item.get("acml_vol") or 0) if not ticker or not name: continue if any(k in name for k in SKIP_KEYWORDS): continue stocks_with_vol.append((ticker, name, volume, market_name)) count += 1 logger.info(" %s: %d개 종목 조회", market_name, count) except Exception as exc: logger.error("❌ %s 종목 조회 실패: %s", market_name, exc) stocks_with_vol.sort(key=lambda x: x[2], reverse=True) top_with_mkt = stocks_with_vol[:limit] top_stocks = [(code, name) for code, name, _, _mkt in top_with_mkt] kospi_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSPI") kosdaq_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSDAQ") logger.info("✅ 선정 %d개 (KOSPI %d / KOSDAQ %d)", len(top_stocks), kospi_n, kosdaq_n) if len(top_stocks) < MIN_STOCK_COUNT: if settings.KIS_MODE == "virtual": # 모의투자 서버는 KOSDAQ 거래량 순위 미지원 — 조회된 종목으로 진행 logger.warning( "⚠️ 모의투자 서버 종목 수 제한: %d개 (기준 %d개). " "KOSDAQ 거래량 순위 미지원으로 인한 정상 동작 — 배치 계속 진행합니다.", len(top_stocks), MIN_STOCK_COUNT ) else: raise RuntimeError( f"종목 조회 결과 너무 적음: {len(top_stocks)}개 (최소 {MIN_STOCK_COUNT}개 필요). " "KIS API 오류 또는 장 휴장 여부를 확인하세요." ) return top_stocks # ------------------------------------------------------------------- # 3. 단일 종목 분석 (순수 분석만 — DB 저장 없음) # ------------------------------------------------------------------- async def analyze_one_stock( idx: int, total: int, stock_code: str, stock_name: str, ) -> tuple[str, Optional[StockRecommendation]]: """종목 분석만 수행하고 결과 반환. DB 저장은 Streaming 저장기에서 처리. Returns: (status, result) status: "ok" | "no_chart" | "price_filter" | "error" """ try: logger.info("[%d/%d] 분석 중: %s (%s)", idx, total, stock_code, stock_name) result = await analyze_stock_for_batch(stock_code, stock_name) if not result: logger.info("샘 차트없음 %s (%s)", stock_code, stock_name) return "no_chart", None # 5,000원 이하 저가주 제외 (CLAUDE.md 5절 가격 필터) if not passes_price_filter(result): logger.info("샘 가격필터 %s: %.0f원 < 5,000원", stock_code, result.current_price) return "price_filter", None ml_info = f", ml_score={result.ml_score}" if result.ml_score is not None else "" logger.info("✅ 분석완료 %s (%s): score=%s%s", stock_code, stock_name, result.score, ml_info) return "ok", result except Exception as exc: logger.error("❌ Failed %s: %s", stock_code, exc) return "error", None # ------------------------------------------------------------------- # 4. Streaming 저장기 — 분석 완료 즉시 BULK_SAVE_CHUNK 단위 UPSERT # ------------------------------------------------------------------- async def _streaming_saver( queue: asyncio.Queue, analysis_date: date, ) -> dict: """큐에서 분석 결과를 꺼내 BULK_SAVE_CHUNK(20)개마다 즉시 DB UPSERT. 전체 메모리 적재 없이 분석 완료 즉시 저장하여 메모리 사용 최소화. Returns: stats dict {"ok", "no_chart", "price_filter", "error", "saved"} """ buffer: list[tuple[str, str, StockRecommendation]] = [] stats = {"ok": 0, "no_chart": 0, "price_filter": 0, "error": 0, "saved": 0} try: while True: item = await queue.get() if item is None: # sentinel: 분석 완료 break status, code, name, result = item stats[status] = stats.get(status, 0) + 1 if status == "ok" and result: buffer.append((code, name, result)) # 버퍼가 청크 크기에 도달하면 즉시 저장 if len(buffer) >= BULK_SAVE_CHUNK: async with AsyncSessionLocal() as session: saved = await bulk_upsert_analysis(session, buffer, analysis_date) await session.commit() stats["saved"] += saved logger.info("💾 DB 저장: %d건 누적", stats["saved"]) buffer.clear() # 나머지 잔여분 저장 if buffer: async with AsyncSessionLocal() as session: saved = await bulk_upsert_analysis(session, buffer, analysis_date) await session.commit() stats["saved"] += saved logger.info("💾 DB 저장(잔여): %d건 누적", stats["saved"]) except Exception as exc: # 🔧 [fix] saver_task 예외 시 큐를 드레인하여 분석 태스크 deadlock 방지 logger.error("❌ _streaming_saver 예외 발생 — 큐 드레인 후 종료: %s", exc, exc_info=True) # 큐에 남은 항목 모두 소비하여 analyze_one의 queue.put() 블로킹 해제 while not queue.empty(): try: queue.get_nowait() except asyncio.QueueEmpty: break return stats # ------------------------------------------------------------------- # 6. 메인 배치 # ------------------------------------------------------------------- async def batch_analyze_all_stocks(limit: int = None): """전체 종목 배치 분석 실행 순서: 1. 시장 레짐 판별 (KOSPI 지수 기준) 2. 시가총액 상위 500개 종목 조회 3. 세마포어로 BATCH_CONCURRENCY(10)개씩 병렬 분석 (API 호출만) 4. Streaming: 분석 완료 즉시 BULK_SAVE_CHUNK(20)개 단위 DB UPSERT 5. DB 스캐너 실행 (추가 API 호출 없음) 6. 워치리스트 초기화 — 스캐너 신호 종목 우선 성능 개선 (기존 대비): - DB 트랜잭션: DELETE+INSERT race-condition → 원자적 UPSERT - 메모리: 전체 결과 적재 → Streaming(분석 즉시 저장) - DB 엔진: 스크립트 자체 생성 → 공통 모듈 공유 """ start_time = datetime.now(KST) analysis_date = datetime.now(KST).date() logger.info("🚀 Starting daily batch analysis for %s", analysis_date) logger.info("⚙️ 동시 처리: %d종목 | 벌크저장 단위: %d개 | API 세마포어: %dreq", BATCH_CONCURRENCY, BULK_SAVE_CHUNK, settings.API_MAX_CONCURRENT_REQUESTS) # Step 1: 시장 레짐 판별 regime = await detect_and_log_market_regime() # Step 2: 종목 목록 조회 (모듈 레벨 싱글톤 재사용, 불필요한 새 인스턴스 생성 금지) all_stocks = await get_all_tradable_stocks(korea_investment_api, limit=500) filtered_stocks = all_stocks[:limit] if limit else all_stocks total = len(filtered_stocks) logger.info("📊 분석 대상: %d개 (시장 레짐: %s)", total, regime) # Step 3 + 4: 병렬 분석 & Streaming 저장 # 분석 태스크 → queue → _streaming_saver (동시 실행) queue: asyncio.Queue = asyncio.Queue(maxsize=BULK_SAVE_CHUNK * 3) semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) async def analyze_one(i: int, code: str, name: str): try: async with semaphore: status, result = await analyze_one_stock(i, total, code, name) await queue.put((status, code, name, result)) except Exception as exc: logger.error("❌ analyze_one unexpected error %s: %s", code, exc) await queue.put(("error", code, name, None)) # 🔧 [fix] saver_task 예외 감지: asyncio.wait로 saver_task 비정상 종료 시 조기 감지 saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date)) # 분석 태스크 전체 실행 analysis_gather = asyncio.gather( *[analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)], return_exceptions=True, ) # saver_task가 예외로 죽으면 analysis_gather도 조기 취소 done, pending = await asyncio.wait( [asyncio.ensure_future(analysis_gather), saver_task], return_when=asyncio.FIRST_EXCEPTION, ) # saver_task 예외 발생 시 분석 태스크 취소 처리 for task in pending: task.cancel() try: await task except (asyncio.CancelledError, Exception): pass # sentinel 전송 → 저장기 종료 (정상 흐름에서도 항상 실행) if not saver_task.done(): await queue.put(None) stats = await saver_task if not saver_task.cancelled() else {} ok_count = stats.get("ok", 0) no_chart = stats.get("no_chart", 0) price_flt = stats.get("price_filter", 0) err_count = stats.get("error", 0) save_count = stats.get("saved", 0) # Step 5: DB 스캐너 (배치 완료 후 추가 API 호출 없이) scanner_tickers = await run_db_scanner(analysis_date) # Step 6: 워치리스트 초기화 — 스캐너 신호 종목 우선 try: from app.services.watchlist import initialize_watchlist async with AsyncSessionLocal() as session: count = await initialize_watchlist( session, analysis_date, scanner_tickers=scanner_tickers ) logger.info("📋 워치리스트 초기화 완료: %d개 (스캐너 신호 %d개 포함)", count, len(scanner_tickers)) except Exception as exc: logger.warning("⚠️ 워치리스트 초기화 실패 (non-critical): %s", exc) # 결과 요약 duration = (datetime.now(KST) - start_time).total_seconds() api_stats = korea_investment_api.get_monitor_stats() logger.info("=" * 60) logger.info("✅ Batch analysis completed!") logger.info("🌏 시장 레짐: %s", regime) logger.info("📊 성공: %d / 전체: %d | DB저장: %d건", ok_count, total, save_count) logger.info(" ├─ 차트없음: %d건", no_chart) logger.info(" ├─ 가격필터: %d건", price_flt) logger.info(" ├─ 오류: %d건", err_count) logger.info("🔍 DB 스캐너 신호: %d개", len(scanner_tickers)) logger.info("⏱️ %.1f초 (%.1f분)", duration, duration / 60) logger.info("📡 API 통계 — 호출:%d 오류:%d(%.1f%%) 평균:%.2f초", api_stats['total_calls'], api_stats['total_errors'], api_stats['error_rate'] * 100, api_stats['avg_duration']) logger.info("=" * 60) # ------------------------------------------------------------------- # Entry Point # ------------------------------------------------------------------- async def main(): import argparse parser = argparse.ArgumentParser(description='Daily Stock Analysis Batch') parser.add_argument('--limit', type=int, help='Limit number of stocks (for testing)') parser.add_argument('--test', action='store_true', help='Test mode (10 stocks)') args = parser.parse_args() if args.test: logger.info("🧪 TEST MODE: 10개 종목") await batch_analyze_all_stocks(limit=10) elif args.limit: await batch_analyze_all_stocks(limit=args.limit) else: await batch_analyze_all_stocks() if __name__ == "__main__": # 🔧 [fix] basicConfig를 __main__ 블록으로 이동 # 인프로세스 스케줄러(main.py lifespan)에서 import될 때 로그 설정 충돌 방지 from app.core.logging_config import KSTFormatter _fmt = KSTFormatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') _handler = logging.StreamHandler() _handler.setFormatter(_fmt) logging.basicConfig(level=logging.INFO, handlers=[_handler]) asyncio.run(main())