코드리뷰 수정

backend/app/main.py

@@ -50,48 +50,54 @@ logger = logging.getLogger(__name__)
 
 
 def _normalize_recommendations(rec: dict) -> dict:
-    """사로 다른 포맷의 추청 결과를 recommendations_by_rule 포맷으로 정규화.
+    """서로 다른 포맷의 추천 결과를 recommendations_by_rule 포맷으로 정규화.
 
     - get_rule_based_recommendations_realtime() → {"recommendations": [...], ...}
     - get_rule_based_recommendations_from_db()   → {"recommendations_by_rule": {...}, ...}
 
     둘 다 recommendations_by_rule 키를 보장한 상태로 돌려준다.
     Redis 캐시 데이터와 DB 데이터 모두 이 함수를 통해 정규화.
+    변환 실패 시 원본 rec를 그대로 반환 (런타임 에러 방지).
     """
     if not rec:
+        return rec or {}
+
+    try:
+        # Case 1: Already normalized (dict)
+        if isinstance(rec.get("recommendations_by_rule"), dict):
+            return rec
+
+        # Determine the source of the recommendation list
+        raw_list = []
+        if "recommendations" in rec and isinstance(rec["recommendations"], list):
+            raw_list = rec["recommendations"]
+        elif "recommendations_by_rule" in rec and isinstance(rec["recommendations_by_rule"], list):
+            # This is the case that causes the bug
+            raw_list = rec["recommendations_by_rule"]
+
+        if not raw_list:
+            # if there is no list to process, make sure the output has a dict
+            rec["recommendations_by_rule"] = {}
+            return rec
+
+        # Group the list by rule
+        by_rule: dict = {}
+        for item in raw_list:
+            rule = (
+                item.get("rule_name")
+                or item.get("signal")
+                or item.get("strategy")
+                or "general"
+            )
+            by_rule.setdefault(rule, []).append(item)
+
+        # Return the original dict with the new/replaced 'recommendations_by_rule'
+        return {**rec, "recommendations_by_rule": by_rule}
+
+    except Exception as e:
+        logger.warning(f"⚠️ _normalize_recommendations 변환 실패, 원본 반환: {e}")
         return rec
 
-    # Case 1: Already normalized (dict)
-    if isinstance(rec.get("recommendations_by_rule"), dict):
-        return rec
-
-    # Determine the source of the recommendation list
-    raw_list = []
-    if "recommendations" in rec and isinstance(rec["recommendations"], list):
-        raw_list = rec["recommendations"]
-    elif "recommendations_by_rule" in rec and isinstance(rec["recommendations_by_rule"], list):
-        # This is the case that causes the bug
-        raw_list = rec["recommendations_by_rule"]
-
-    if not raw_list:
-        # if there is no list to process, make sure the output has a dict
-        rec["recommendations_by_rule"] = {}
-        return rec
-
-    # Group the list by rule
-    by_rule: dict = {}
-    for item in raw_list:
-        rule = (
-            item.get("rule_name")
-            or item.get("signal")
-            or item.get("strategy")
-            or "general"
-        )
-        by_rule.setdefault(rule, []).append(item)
-
-    # Return the original dict with the new/replaced 'recommendations_by_rule'
-    return {**rec, "recommendations_by_rule": by_rule}
-
 
 async def batch_analysis_scheduler():
     """10분 간격 배치 분석 스케줄러.
@@ -99,57 +105,92 @@ async def batch_analysis_scheduler():
     장중 10분마다 KIS API로 종목 분석 후 stock_analysis DB에 upsert.
     분석 완료 후 Redis에 recommendations:latest 캐시를 저장해
     auto_trade_scheduler가 DB 조회 없이 바로 매매 판단에 활용할 수 있도록 한다.
+
+    개선:
+    - 15:30 이후 장 종료 감지 → 익일 08:50까지 자동 중단
+    - 분석 실패 시 exponential backoff 재시도 (최대 3회: 10s/20s/40s)
     """
     await asyncio.sleep(15)  # 부팅 안정화 (auto_trade_scheduler보다 늦게 시작)
     from app.api.endpoints.recommendations import get_rule_based_recommendations_realtime
     from app.core.redis import setex_json, TTL_RECOMMENDATIONS, TTL_ANALYSIS
+    from datetime import timedelta as _td
 
     interval = int(getattr(settings, "BATCH_ANALYSIS_INTERVAL", 600))  # 기본 10분
+    _KST = pytz.timezone("Asia/Seoul")
     logger.info(f"📊 Batch analysis scheduler started (every {interval}s)")
     try:
         while True:
+            now_kst = pytz.datetime.datetime.now(_KST)
+
+            # 15:30 이후 → 장 종료, 익일 08:50까지 대기
+            if now_kst.hour > 15 or (now_kst.hour == 15 and now_kst.minute >= 30):
+                tomorrow_start = (
+                    _KST.localize(
+                        pytz.datetime.datetime(now_kst.year, now_kst.month, now_kst.day, 8, 50, 0)
+                    ) + _td(days=1)
+                )
+                sleep_sec = max((tomorrow_start - now_kst).total_seconds(), 3600)
+                logger.info(f"⏸️ [Batch] 장 종료 (15:30 이후). {sleep_sec / 3600:.1f}시간 후 재개")
+                await asyncio.sleep(sleep_sec)
+                continue
+
             if not is_market_open():
                 await asyncio.sleep(1800)
                 continue
-            try:
-                logger.info("🔄 [Batch] 10-min analysis cycle starting...")
-                result = await get_rule_based_recommendations_realtime()
-                count = len(result.get("recommendations", [])) if result else 0
-                logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB")
 
-                # ── Redis 캐시 업데이트 ─────────────────────────────────────────
-                if result:
-                    # 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환)
-                    normalized = _normalize_recommendations(result)
-                    ok = await setex_json(
-                        "recommendations:latest", TTL_RECOMMENDATIONS, normalized
-                    )
-                    if ok:
-                        rec_by_rule = normalized.get('recommendations_by_rule', [])
-                        rec_count = len(rec_by_rule) if isinstance(rec_by_rule, (list, dict)) else 0
-                        logger.info(
-                            f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest "
-                            f"(rule 수: {rec_count}, "
-                            f"TTL: {TTL_RECOMMENDATIONS}s)"
+            # ── exponential backoff 재시도 (최대 3회) ────────────────────────
+            max_retries = 3
+            for attempt in range(max_retries):
+                try:
+                    suffix = f" (재시도 {attempt + 1}/{max_retries})" if attempt > 0 else ""
+                    logger.info(f"🔄 [Batch] 10-min analysis cycle starting...{suffix}")
+                    result = await get_rule_based_recommendations_realtime()
+                    count = len(result.get("recommendations", [])) if result else 0
+                    logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB")
+
+                    # ── Redis 캐시 업데이트 ─────────────────────────────────────────
+                    if result:
+                        # 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환)
+                        normalized = _normalize_recommendations(result)
+                        ok = await setex_json(
+                            "recommendations:latest", TTL_RECOMMENDATIONS, normalized
                         )
-                        
-                    # ===== 개별 종목별 캐시 (병렬 처리 적용) =====
-                    raw_list = result.get("recommendations", [])
-                    cache_tasks = []
-                    
-                    for item in raw_list:
-                        code = item.get("stock_code") or item.get("code")
-                        if code:
-                            cache_tasks.append(setex_json(f"analysis:{code}", TTL_ANALYSIS, item))
-                            
-                    if cache_tasks:
-                        await asyncio.gather(*cache_tasks, return_exceptions=True)
-                        logger.info(f"💾 [Batch] 개별 종목 분석 데이터 캐시 갱신 완료: {len(cache_tasks)}건 (TTL: {TTL_ANALYSIS}s)")
-                    # ===============================================
-                # ───────────────────────────────────────────────────────
+                        if ok:
+                            rec_by_rule = normalized.get('recommendations_by_rule', [])
+                            rec_count = len(rec_by_rule) if isinstance(rec_by_rule, (list, dict)) else 0
+                            logger.info(
+                                f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest "
+                                f"(rule 수: {rec_count}, "
+                                f"TTL: {TTL_RECOMMENDATIONS}s)"
+                            )
+
+                        # ===== 개별 종목별 캐시 (병렬 처리 적용) =====
+                        raw_list = result.get("recommendations", [])
+                        cache_tasks = []
+
+                        for item in raw_list:
+                            code = item.get("stock_code") or item.get("code")
+                            if code:
+                                cache_tasks.append(setex_json(f"analysis:{code}", TTL_ANALYSIS, item))
+
+                        if cache_tasks:
+                            await asyncio.gather(*cache_tasks, return_exceptions=True)
+                            logger.info(f"💾 [Batch] 개별 종목 분석 데이터 캐시 갱신 완료: {len(cache_tasks)}건 (TTL: {TTL_ANALYSIS}s)")
+                        # ===============================================
+                    # ───────────────────────────────────────────────────────
+                    break  # 성공 시 재시도 루프 탈출
+
+                except Exception as e:
+                    wait = 10 * (2 ** attempt)  # 10s → 20s → 40s
+                    if attempt < max_retries - 1:
+                        logger.warning(
+                            f"⚠️ [Batch] 분석 실패 ({attempt + 1}/{max_retries}), "
+                            f"{wait}초 후 재시도: {e}"
+                        )
+                        await asyncio.sleep(wait)
+                    else:
+                        logger.error(f"❌ Batch analysis error (최대 재시도 초과): {e}", exc_info=True)
 
-            except Exception as e:
-                logger.error(f"❌ Batch analysis error: {e}", exc_info=True)
             await asyncio.sleep(interval)
     except asyncio.CancelledError:
         logger.info("⏹️ Batch analysis scheduler stopped")
@@ -381,16 +422,26 @@ async def krx_data_collector_scheduler():
                 await asyncio.sleep(3600)
                 continue
 
-            # 수집 실행
-            try:
-                logger.info("📥 [KRX] 일봉 데이터 증분 수집 시작 (update mode)...")
-                from collect_krx_data import collect_all
-                await collect_all(update_mode=True)
-                _last_collected = today
-                logger.info("✅ [KRX] 일봉 데이터 수집 완료")
-            except Exception as e:
-                logger.error(f"❌ [KRX] 데이터 수집 실패: {e}", exc_info=True)
-                await asyncio.sleep(1800)  # 실패 시 30분 후 재시도
+            # 수집 실행 (최대 3회 재시도, 30분 간격)
+            max_krx_retries = 3
+            for krx_attempt in range(max_krx_retries):
+                try:
+                    suffix = f" (재시도 {krx_attempt + 1}/{max_krx_retries})" if krx_attempt > 0 else ""
+                    logger.info(f"📥 [KRX] 일봉 데이터 증분 수집 시작 (update mode){suffix}...")
+                    from collect_krx_data import collect_all
+                    await collect_all(update_mode=True)
+                    _last_collected = today
+                    logger.info("✅ [KRX] 일봉 데이터 수집 완료")
+                    break
+                except Exception as e:
+                    if krx_attempt < max_krx_retries - 1:
+                        logger.warning(
+                            f"⚠️ [KRX] 수집 실패 ({krx_attempt + 1}/{max_krx_retries}), "
+                            f"30분 후 재시도: {e}"
+                        )
+                        await asyncio.sleep(1800)
+                    else:
+                        logger.error(f"❌ [KRX] 데이터 수집 최대 재시도 초과: {e}", exc_info=True)
 
     except asyncio.CancelledError:
         logger.info("⏹️ KRX data collector scheduler stopped")
@@ -405,6 +456,10 @@ async def lifespan(app: FastAPI):
     logger.info(f"KIS Base URL: {settings.KOREA_INVESTMENT_API_BASE_URL}")
     logger.info(f"KIS Account: {settings.KOREA_INVESTMENT_ACCOUNT_NUMBER}")
 
+    # ── 필수 환경변수 검증 (조기 경보) ──────────────────────────────────────────
+    for warn_msg in settings.validate_required():
+        logger.warning(f"⚠️ [Config] {warn_msg}")
+
     # DB 테이블 자동 생성
     # 컬럼 추가/테이블 생성 DDL 은 Alembic migration 으로 관리됨.
     # 배포 전 `cd backend && alembic upgrade head` 를 실행할 것.
@@ -595,6 +650,21 @@ async def log_requests(request: Request, call_next):
 app.include_router(api_router, prefix="/api/v1")
 
 
+# ── 전역 예외 핸들러 ────────────────────────────────────────────────────────────
+from fastapi import Request as _Request
+from fastapi.responses import JSONResponse as _JSONResponse
+
+
+@app.exception_handler(Exception)
+async def global_exception_handler(_req: _Request, exc: Exception) -> _JSONResponse:
+    """처리되지 않은 예외를 500 JSON으로 변환해 일관된 에러 응답을 보장한다."""
+    logger.error(f"Unhandled exception: {_req.method} {_req.url} — {exc}", exc_info=True)
+    return _JSONResponse(
+        status_code=500,
+        content={"detail": "내부 서버 오류가 발생했습니다. 잠시 후 다시 시도해 주세요."},
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Auto Stock Trading System API", "version": "1.0.0", "status": "running"}
@@ -602,7 +672,34 @@ async def root():
 
 @app.get("/health")
 async def health_check():
-    return {"status": "healthy"}
+    """DB / Redis 연결 상태를 포함한 헬스체크."""
+    from sqlalchemy import text as _sa_text
+    from app.core.redis import get_redis
+
+    db_ok = False
+    redis_ok = False
+
+    try:
+        async with engine.connect() as conn:
+            await conn.execute(_sa_text("SELECT 1"))
+        db_ok = True
+    except Exception:
+        pass
+
+    try:
+        r = get_redis()
+        if r:
+            await r.ping()
+            redis_ok = True
+    except Exception:
+        pass
+
+    status = "healthy" if (db_ok and redis_ok) else "degraded"
+    return {
+        "status": status,
+        "db": "ok" if db_ok else "error",
+        "redis": "ok" if redis_ok else "error",
+    }
 
 
 @app.websocket("/ws/{client_id}")
@@ -621,7 +718,17 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
     await websocket_manager.connect(websocket, client_id)
     try:
         while True:
-            data = await websocket.receive_text()
+            try:
+                data = await asyncio.wait_for(websocket.receive_text(), timeout=60.0)
+            except asyncio.TimeoutError:
+                # 60초 동안 메시지 없음 → ping 전송으로 연결 유지 확인
+                try:
+                    await websocket_manager.send_personal_message(
+                        json.dumps({"type": "ping"}), client_id
+                    )
+                except Exception:
+                    break
+                continue
             try:
                 msg = json.loads(data)
             except (json.JSONDecodeError, AttributeError):