fix: batch_daily_analysis 리뷰 이슈 수정
- [🔴] saver_task 예외 시 deadlock 방지: asyncio.wait로 saver_task 예외 감지 - [🔴] logging.basicConfig를 __main__ 블록으로 이동 (인프로세스 실행 시 충돌 방지) - [🟡] get_all_tradable_stocks: 최소 종목 수(50개) 미만 시 RuntimeError 발생 - [🟠] detect_and_log_market_regime: df=None일 때 len(df) 호출 안전하게 수정
This commit is contained in:
@@ -36,10 +36,8 @@ from app.services.analysis_db_service import bulk_upsert_analysis, passes_price_
|
||||
from app.api.endpoints.recommendations import analyze_stock_for_batch, StockRecommendation
|
||||
import logging
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
# 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출
|
||||
# 인프로세스 스케줄러(main.py)로 실행될 때 로그 설정 충돌 방지
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
@@ -48,6 +46,9 @@ logger = logging.getLogger(__name__)
|
||||
BATCH_CONCURRENCY = 10 # API 세마포어 허용 범위 내 동시 처리 수
|
||||
BULK_SAVE_CHUNK = 20 # Streaming 저장 단위 (큐에 20개 쌓이면 즉시 UPSERT)
|
||||
|
||||
# 종목 목록 조회 실패 시 배치를 중단할 최소 종목 수
|
||||
MIN_STOCK_COUNT = 50
|
||||
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# 1. 시장 레짐 판별 (KOSPI 기준)
|
||||
@@ -68,8 +69,11 @@ async def detect_and_log_market_regime() -> str:
|
||||
|
||||
# KOSPI 지수 코드 1001
|
||||
df = pykrx_stock.get_index_ohlcv_by_date(start, today, "1001")
|
||||
if df is None or len(df) < 205:
|
||||
logger.warning("⚠️ KOSPI 데이터 부족(행 %d) — 레짐 판별 스킵", 0 if df is None else len(df))
|
||||
|
||||
# 🔧 [fix] df=None일 때 len(df) 직접 호출 방지
|
||||
df_len = len(df) if df is not None else 0
|
||||
if df_len < 205:
|
||||
logger.warning("⚠️ KOSPI 데이터 부족(행 %d) — 레짐 판별 스킵", df_len)
|
||||
return "UNKNOWN"
|
||||
|
||||
kospi_closes = df["종가"].tolist()
|
||||
@@ -94,6 +98,9 @@ async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500)
|
||||
|
||||
ETF / 레버리지 / 스팭 등 특수종목 자동 제외.
|
||||
KOSPI + KOSDAQ 합산 시가총액 상위 limit개 반환.
|
||||
|
||||
Raises:
|
||||
RuntimeError: 조회 결과가 MIN_STOCK_COUNT(50) 미만인 경우 배치 중단
|
||||
"""
|
||||
logger.info("Fetching top %d stocks by market cap from pykrx...", limit)
|
||||
try:
|
||||
@@ -130,8 +137,18 @@ async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500)
|
||||
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)
|
||||
|
||||
# 🔧 [fix] 최소 종목 수 검증 — 너무 적으면 잘못된 데이터로 배치 실행 방지
|
||||
if len(top_stocks) < MIN_STOCK_COUNT:
|
||||
raise RuntimeError(
|
||||
f"종목 조회 결과 너무 적음: {len(top_stocks)}개 (최소 {MIN_STOCK_COUNT}개 필요). "
|
||||
"pykrx API 오류 또는 장 휴장 여부를 확인하세요."
|
||||
)
|
||||
|
||||
return top_stocks
|
||||
|
||||
except RuntimeError:
|
||||
raise # 최소 종목 수 미달 — 상위로 전파하여 배치 중단
|
||||
except Exception as exc:
|
||||
logger.error("❌ 종목 목록 조회 실패: %s", exc, exc_info=True)
|
||||
return []
|
||||
@@ -191,6 +208,7 @@ async def _streaming_saver(
|
||||
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: 분석 완료
|
||||
@@ -219,6 +237,16 @@ async def _streaming_saver(
|
||||
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
|
||||
|
||||
|
||||
@@ -272,16 +300,34 @@ async def batch_analyze_all_stocks(limit: int = None):
|
||||
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))
|
||||
|
||||
# 분석 태스크 전체 실행
|
||||
tasks = [analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)]
|
||||
await asyncio.gather(*tasks, return_exceptions=True)
|
||||
analysis_gather = asyncio.gather(
|
||||
*[analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
# sentinel 전송 → 저장기 종료
|
||||
# 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
|
||||
|
||||
stats = await saver_task if not saver_task.cancelled() else {}
|
||||
|
||||
ok_count = stats.get("ok", 0)
|
||||
no_chart = stats.get("no_chart", 0)
|
||||
@@ -344,4 +390,10 @@ async def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 🔧 [fix] basicConfig를 __main__ 블록으로 이동
|
||||
# 인프로세스 스케줄러(main.py lifespan)에서 import될 때 로그 설정 충돌 방지
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
asyncio.run(main())
|
||||
|
||||
Reference in New Issue
Block a user