refactor: MCP 공통 함수/httpx 싱글톤, CORS 수정, ML 이벤트루프 블로킹 해소 (#4~#7)

2026-02-24 18:25:36 +09:00
parent 24f3082693
commit b36b4a381f
6 changed files with 333 additions and 308 deletions
--- a/backend/app/api/endpoints/mcp.py
+++ b/backend/app/api/endpoints/mcp.py
@@ -1,12 +1,117 @@
+"""
+MCP 프록시 엔드포인트 + 공통 리스크 평가 유틸리티
+
+이슈 4 - request_mcp_risk_assessment():
+    auto/day 스케줄러 양쪽에서 재사용 가능한 공통 RISK_ASSESSMENT 함수.
+    호출부는 trade_plan / portfolio / signals 세 dict를 넘긴다.
+
+이슈 6 - httpx.AsyncClient 싱글톤:
+    요청마다 클라이언트를 생성/소멸하지 않는다.
+    lifespan에서 init_http_client() → 사용 → close_http_client() 순서로 호출.
+"""
 from fastapi import APIRouter, HTTPException
-from typing import Dict, Any
+from typing import Dict, Any, Optional, Tuple
 from pydantic import BaseModel
 import httpx
+import logging
+
 from app.core.config import settings

+logger = logging.getLogger(__name__)
 router = APIRouter()


+# ─── httpx 싱글톤 클라이언트 ──────────────────────────────────────────
+_http_client: Optional[httpx.AsyncClient] = None
+
+
+async def init_http_client() -> None:
+    """lifespan 시작 시 한 번만 호출. keepalive 연결 풀 설정."""
+    global _http_client
+    _http_client = httpx.AsyncClient(
+        timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0),
+        limits=httpx.Limits(max_keepalive_connections=5, max_connections=10),
+    )
+    logger.info("✅ MCP httpx.AsyncClient initialized")
+
+
+async def close_http_client() -> None:
+    """lifespan 종료 시 한 번만 호출."""
+    global _http_client
+    if _http_client is not None:
+        await _http_client.aclose()
+        _http_client = None
+        logger.info("🔒 MCP httpx.AsyncClient closed")
+
+
+def get_http_client() -> httpx.AsyncClient:
+    """공유 클라이언트 반환. 초기화 전 접근 시 RuntimeError."""
+    if _http_client is None:
+        raise RuntimeError(
+            "http_client not initialized — call init_http_client() in lifespan"
+        )
+    return _http_client
+
+
+# ─── 공통 리스크 평가 함수 ────────────────────────────────────────────
+async def request_mcp_risk_assessment(
+    trade_plan: Dict[str, Any],
+    portfolio: Dict[str, Any],
+    signals: Dict[str, Any],
+) -> Tuple[bool, str]:
+    """MCP RISK_ASSESSMENT 호출 — auto/day 스케줄러 공통 진입점.
+
+    Args:
+        trade_plan:  주문 계획 (action, strategy, stock_code, quantity, price 등)
+        portfolio:   현재 포트폴리오 요약 (보유 종목 목록, 총 평가금액 등)
+        signals:     진입 신호 정보 (strategy, rules, indicators 등)
+
+    Returns:
+        (approve: bool, reason: str)
+        - approve=True  → 주문 실행
+        - approve=False → 주문 스킵
+        - MCP 장애 / 타임아웃 → fail-open (approve=True, 매매 중단 방지)
+    """
+    try:
+        client = get_http_client()
+    except RuntimeError:
+        logger.warning("⚠️ http_client 미초기화 — fail-open")
+        return True, "client_not_initialized_fail_open"
+
+    try:
+        resp = await client.post(
+            f"{settings.MCP_SERVER_URL}/analyze",
+            json={
+                "analysis_type": "RISK_ASSESSMENT",
+                "data": {
+                    "trade_plan": trade_plan,
+                    "portfolio": portfolio,
+                    "signals": signals,
+                },
+            },
+            timeout=15.0,
+        )
+        if resp.status_code == 200:
+            result = resp.json()
+            # 응답: {"result": {"decision": "approve"|"deny", "reason": "..."}}
+            payload = result.get("result", result)
+            decision = str(payload.get("decision", "approve")).lower()
+            reason = str(payload.get("reason", ""))
+            if "deny" in decision or "reject" in decision:
+                return False, reason
+            return True, reason
+        else:
+            logger.warning(f"⚠️ MCP RISK_ASSESSMENT HTTP {resp.status_code} — fail-open")
+            return True, f"mcp_http_{resp.status_code}_fail_open"
+    except httpx.TimeoutException:
+        logger.warning("⚠️ MCP RISK_ASSESSMENT 타임아웃 — fail-open")
+        return True, "mcp_timeout_fail_open"
+    except Exception as e:
+        logger.warning(f"⚠️ MCP RISK_ASSESSMENT 오류: {e} — fail-open")
+        return True, f"mcp_error_fail_open: {e}"
+
+
+# ─── 모델 ──────────────────────────────────────────────────────────────
 class MCPToolRequest(BaseModel):
    tool_name: str
    parameters: Dict[str, Any]
@@ -17,59 +122,53 @@ class MCPAnalysisRequest(BaseModel):
    data: Dict[str, Any]


+# ─── 엔드포인트 ────────────────────────────────────────────────────────
@router.get("/status")
 async def get_mcp_status():
    """MCP 서버 상태 확인"""
    try:
-        async with httpx.AsyncClient() as client:
-            response = await client.get(f"{settings.MCP_SERVER_URL}/health", timeout=2.0)
-            
-            if response.status_code == 200:
-                health_data = response.json()
-                return {
-                    "status": "connected",
-                    "mcp_server_url": settings.MCP_SERVER_URL,
-                    "claude_configured": health_data.get("claude_configured", False),
-                    "ai_provider": health_data.get("ai_provider", "unknown"),
-                    "details": health_data
-                }
-            else:
-                return {
-                    "status": "error",
-                    "message": "MCP server responded with error"
-                }
-    
+        client = get_http_client()
+    except RuntimeError:
+        return {"status": "disconnected", "message": "http_client not initialized"}
+    try:
+        response = await client.get(f"{settings.MCP_SERVER_URL}/health", timeout=2.0)
+        if response.status_code == 200:
+            health_data = response.json()
+            return {
+                "status": "connected",
+                "mcp_server_url": settings.MCP_SERVER_URL,
+                "claude_configured": health_data.get("claude_configured", False),
+                "ai_provider": health_data.get("ai_provider", "unknown"),
+                "details": health_data,
+            }
+        return {"status": "error", "message": "MCP server responded with error"}
    except Exception as e:
-        return {
-            "status": "disconnected",
-            "message": str(e)
-        }
+        return {"status": "disconnected", "message": str(e)}


@router.post("/tools/call")
 async def call_mcp_tool(request: MCPToolRequest):
    """MCP 도구 호출"""
    try:
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                f"{settings.MCP_SERVER_URL}/tools/call",
-                json={
-                    "tool_name": request.tool_name,
-                    "parameters": request.parameters
-                },
-                timeout=30.0
-            )
-            
-            if response.status_code == 200:
-                return response.json()
-            else:
-                raise HTTPException(
-                    status_code=response.status_code,
-                    detail=f"MCP tool call failed: {response.text}"
-                )
-    
+        client = get_http_client()
+    except RuntimeError as e:
+        raise HTTPException(status_code=503, detail=str(e))
+    try:
+        response = await client.post(
+            f"{settings.MCP_SERVER_URL}/tools/call",
+            json={"tool_name": request.tool_name, "parameters": request.parameters},
+            timeout=30.0,
+        )
+        if response.status_code == 200:
+            return response.json()
+        raise HTTPException(
+            status_code=response.status_code,
+            detail=f"MCP tool call failed: {response.text}",
+        )
    except httpx.TimeoutException:
        raise HTTPException(status_code=504, detail="MCP server timeout")
+    except HTTPException:
+        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to call MCP tool: {str(e)}")

@@ -78,26 +177,25 @@ async def call_mcp_tool(request: MCPToolRequest):
 async def analyze_with_claude(request: MCPAnalysisRequest):
    """Claude AI를 통한 분석 요청"""
    try:
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                f"{settings.MCP_SERVER_URL}/analyze",
-                json={
-                    "analysis_type": request.analysis_type,
-                    "data": request.data
-                },
-                timeout=60.0
-            )
-            
-            if response.status_code == 200:
-                return response.json()
-            else:
-                raise HTTPException(
-                    status_code=response.status_code,
-                    detail=f"Claude analysis failed: {response.text}"
-                )
-    
+        client = get_http_client()
+    except RuntimeError as e:
+        raise HTTPException(status_code=503, detail=str(e))
+    try:
+        response = await client.post(
+            f"{settings.MCP_SERVER_URL}/analyze",
+            json={"analysis_type": request.analysis_type, "data": request.data},
+            timeout=60.0,
+        )
+        if response.status_code == 200:
+            return response.json()
+        raise HTTPException(
+            status_code=response.status_code,
+            detail=f"Claude analysis failed: {response.text}",
+        )
    except httpx.TimeoutException:
        raise HTTPException(status_code=504, detail="Analysis timeout")
+    except HTTPException:
+        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to analyze: {str(e)}")

@@ -106,16 +204,17 @@ async def analyze_with_claude(request: MCPAnalysisRequest):
 async def list_mcp_tools():
    """사용 가능한 MCP 도구 목록"""
    try:
-        async with httpx.AsyncClient() as client:
-            response = await client.get(f"{settings.MCP_SERVER_URL}/tools", timeout=10.0)
-            
-            if response.status_code == 200:
-                return response.json()
-            else:
-                raise HTTPException(
-                    status_code=response.status_code,
-                    detail="Failed to fetch tools list"
-                )
-    
+        client = get_http_client()
+    except RuntimeError as e:
+        raise HTTPException(status_code=503, detail=str(e))
+    try:
+        response = await client.get(f"{settings.MCP_SERVER_URL}/tools", timeout=10.0)
+        if response.status_code == 200:
+            return response.json()
+        raise HTTPException(
+            status_code=response.status_code, detail="Failed to fetch tools list"
+        )
+    except HTTPException:
+        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to list tools: {str(e)}")
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -76,100 +76,76 @@ async def batch_analysis_scheduler():
        raise


-async def _call_mcp_risk_assessment(rec: dict) -> tuple[bool, str]:
-    """MCP RISK_ASSESSMENT 호출.
-    반환: (approve: bool, reason: str)
-    MCP 서버 장애 / 타임아웃 시 기본 approve=True (fail-open)로 처리.
-    """
-    import httpx
-
-    # 추천 요약 데이터 구성
-    recommendations = rec.get("recommendations_by_rule", {})
-    summary = {
-        "recommendation_count": sum(len(v) for v in recommendations.values()),
-        "rules": list(recommendations.keys()),
-        "top_stocks": [
-            item.get("stock_code") or item.get("code", "")
-            for rule_items in recommendations.values()
-            for item in (rule_items[:3] if isinstance(rule_items, list) else [])
-        ][:10],
-    }
-
-    try:
-        async with httpx.AsyncClient() as client:
-            resp = await client.post(
-                f"{settings.MCP_SERVER_URL}/analyze",
-                json={
-                    "analysis_type": "RISK_ASSESSMENT",
-                    "data": summary,
-                },
-                timeout=15.0,
-            )
-        if resp.status_code == 200:
-            result = resp.json()
-            # MCP 서버 응답 형식: {"decision": "approve"|"deny", "reason": "..."}
-            # 또는 {"result": {"decision": ..., "reason": ...}}
-            payload = result.get("result", result)
-            decision = str(payload.get("decision", "approve")).lower()
-            reason = str(payload.get("reason", ""))
-            if "deny" in decision or "reject" in decision:
-                return False, reason
-            return True, reason
-        else:
-            logger.warning(f"⚠️ MCP RISK_ASSESSMENT HTTP {resp.status_code} — fail-open")
-            return True, "mcp_http_error_fail_open"
-    except httpx.TimeoutException:
-        logger.warning("⚠️ MCP RISK_ASSESSMENT 타임아웃 — fail-open")
-        return True, "mcp_timeout_fail_open"
-    except Exception as e:
-        logger.warning(f"⚠️ MCP RISK_ASSESSMENT 오류: {e} — fail-open")
-        return True, f"mcp_error_fail_open: {e}"
-
-
 async def auto_trade_scheduler():
    """1분 간격 자동매매 스케줄러.
    DB 캐시에서 추천 종목을 읽어 손절/익절 및 신규 매수를 실행한다.
+    주문 직전에 MCP RISK_ASSESSMENT를 호출해 승인된 경우에만 주문한다.
    실제 KIS API 분석은 batch_analysis_scheduler(10분)가 담당한다.
    """
-    # 지연 시작으로 앱 부팅 안정화
    await asyncio.sleep(2)
    from app.api.endpoints.recommendations import (
        get_rule_based_recommendations_from_db,
        execute_auto_trade_background,
    )
    from app.api.endpoints.portfolio import check_and_execute_exit_trades
+    # 공통 MCP 리스크 평가 함수 (이슈 4)
+    from app.api.endpoints.mcp import request_mcp_risk_assessment

    logger.info("⏱️ Auto-trade scheduler started (every 60s, DB mode)")
    try:
        while True:
            try:
-                # 장 운영 시간 체크
                if not is_market_open():
-                    logger.info("💤 Market is closed. Skipping auto-trade cycle. Next check in 30 minutes.")
+                    logger.info("💤 Market is closed. Next check in 30 minutes.")
                    await asyncio.sleep(1800)
                    continue

                async with AsyncSessionLocal() as session:
-                    # 1. 보유 종목 손절/익절 체크 (먼저 실행)
+                    # 1. 보유 종목 손절/익절 체크
                    logger.info("💰 Checking portfolio for stop-loss/take-profit...")
                    try:
                        await check_and_execute_exit_trades(session)
                    except Exception as e:
                        logger.error(f"Exit trade check error: {e}", exc_info=True)

-                    # 2. 자동매매: DB 캐시에서 추천 조회 (빠름 - API 호출 없음)
+                    # 2. 자동매매: DB 캐시에서 추천 조회
                    if settings.AUTO_TRADE_ENABLED:
                        logger.info("🔄 Checking DB recommendations for auto-trade...")
                        rec = await get_rule_based_recommendations_from_db(session)

                        if rec and rec.get("recommendations_by_rule"):
-                            # ── MCP RISK_ASSESSMENT 승인 체크 ──────────────────────
-                            approve, reason = await _call_mcp_risk_assessment(rec)
+                            # ── MCP RISK_ASSESSMENT (공통 함수, 이슈 4) ──────────
+                            recommendations = rec.get("recommendations_by_rule", {})
+                            trade_plan = {
+                                "action": "BUY",
+                                "strategy": "rule_based_batch",
+                                "recommendation_count": sum(
+                                    len(v) for v in recommendations.values()
+                                ),
+                                "rules": list(recommendations.keys()),
+                                "top_stocks": [
+                                    item.get("stock_code") or item.get("code", "")
+                                    for rule_items in recommendations.values()
+                                    for item in (
+                                        rule_items[:3]
+                                        if isinstance(rule_items, list)
+                                        else []
+                                    )
+                                ][:10],
+                            }
+                            portfolio: dict = {}  # 확장 시 실제 포트폴리오 요약 전달
+                            signals = {
+                                "source": "db_recommendations",
+                                "rules_triggered": list(recommendations.keys()),
+                            }
+                            approve, reason = await request_mcp_risk_assessment(
+                                trade_plan, portfolio, signals
+                            )
+                            # ─────────────────────────────────────────────────────
                            if not approve:
                                logger.warning(
                                    f"🛑 MCP RISK_ASSESSMENT DENY → 주문 스킵. 사유: {reason}"
                                )
-                                # deny 사유 로그 저장 (선택적)
                                try:
                                    from app.services.db_worker import enqueue_db_write
                                    await enqueue_db_write({
@@ -180,10 +156,11 @@ async def auto_trade_scheduler():
                                    pass
                            else:
                                if reason:
-                                    logger.info(f"✅ MCP RISK_ASSESSMENT APPROVE. 참고: {reason}")
-                                logger.info(f"📊 Found recommendations, executing auto-trade")
+                                    logger.info(
+                                        f"✅ MCP RISK_ASSESSMENT APPROVE. 참고: {reason}"
+                                    )
+                                logger.info("📊 Found recommendations, executing auto-trade")
                                await execute_auto_trade_background(rec, session)
-                            # ────────────────────────────────────────────────────────
                        else:
                            logger.info("⏭️ Auto-trade skipped: no recommendations in DB")
                    else:
@@ -205,7 +182,7 @@ async def day_trade_scheduler():
    - EOD 강제 청산 (15:20~15:30)
    - 장 외 시간: 30분 슬립
    """
-    await asyncio.sleep(5)  # 부팅 안정화 대기
+    await asyncio.sleep(5)
    from app.api.endpoints.day_trading import run_day_trading_cycle

    logger.info("📈 Day-trade scheduler started (every 30s)")
@@ -238,7 +215,6 @@ async def day_trade_scheduler():
 async def lifespan(app: FastAPI):
    """애플리케이션 시작/종료 시 실행될 코드"""
    logger.info("Starting Auto Stock Trading System...")
-    kis_mode = settings.KIS_MODE.upper()
    logger.info(f"KIS API Mode: {'모의투자 (VIRTUAL)' if settings.is_virtual_trading else '실전투자 (REAL)'}")
    logger.info(f"KIS Base URL: {settings.KOREA_INVESTMENT_API_BASE_URL}")
    logger.info(f"KIS Account: {settings.KOREA_INVESTMENT_ACCOUNT_NUMBER}")
@@ -248,9 +224,8 @@ async def lifespan(app: FastAPI):
        logger.info(f"Connecting to database: {settings.DATABASE_URL.split('@')[1] if '@' in settings.DATABASE_URL else 'unknown'}")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
-            # Ensure trade_history columns exist (self-heal without external migration)
            await conn.execute(text("""
-            ALTER TABLE trade_history 
+            ALTER TABLE trade_history
            ADD COLUMN IF NOT EXISTS total_amount NUMERIC(15, 2),
            ADD COLUMN IF NOT EXISTS entry_reason TEXT,
            ADD COLUMN IF NOT EXISTS exit_reason TEXT,
@@ -264,9 +239,10 @@ async def lifespan(app: FastAPI):
            await conn.execute(text("""
            UPDATE trade_history
            SET total_amount = executed_price * executed_quantity
-            WHERE total_amount IS NULL AND executed_price IS NOT NULL AND executed_quantity IS NOT NULL;
+            WHERE total_amount IS NULL
+              AND executed_price IS NOT NULL
+              AND executed_quantity IS NOT NULL;
            """))
-            # stock_analysis에 ATR / ADX / ML 컬럼 추가 (CLAUDE.md 섹션 6.2)
            await conn.execute(text("""
            ALTER TABLE stock_analysis
            ADD COLUMN IF NOT EXISTS atr_14 NUMERIC(15, 2),
@@ -275,7 +251,6 @@ async def lifespan(app: FastAPI):
            ADD COLUMN IF NOT EXISTS ml_confidence NUMERIC(5, 4),
            ADD COLUMN IF NOT EXISTS ml_signal VARCHAR(10);
            """))
-            # UPSERT를 위한 unique constraint — init-db.sql에 없을 경우 자동 생성
            await conn.execute(text("""
            DO $$ BEGIN
                IF NOT EXISTS (
@@ -289,14 +264,12 @@ async def lifespan(app: FastAPI):
                END IF;
            END $$;
            """))
-            # active_trades에 분할 청산 컬럼 추가 (CLAUDE.md 섹션 7.3)
            await conn.execute(text("""
            ALTER TABLE active_trades
            ADD COLUMN IF NOT EXISTS remaining_pct NUMERIC(5, 2) DEFAULT 1.0,
            ADD COLUMN IF NOT EXISTS highest_since_entry NUMERIC(15, 2),
            ADD COLUMN IF NOT EXISTS strategy_name VARCHAR(100);
            """))
-            # Risk Management 테이블 생성
            await conn.execute(text("""
            CREATE TABLE IF NOT EXISTS trading_sessions (
                id SERIAL PRIMARY KEY,
@@ -309,7 +282,6 @@ async def lifespan(app: FastAPI):
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
            """))
-            # 종목별 연속 손실 ban 영속화 테이블 (CLAUDE.md 섹션 7.4)
            await conn.execute(text("""
            CREATE TABLE IF NOT EXISTS stock_bans (
                stock_code VARCHAR(20) PRIMARY KEY,
@@ -320,7 +292,6 @@ async def lifespan(app: FastAPI):
            """))
        logger.info("✅ Database tables created successfully (and columns verified)")

-        # 연속 손실 ban 상태 복원 (재시작 후에도 유지 — CLAUDE.md 섹션 7.4)
        try:
            from app.services.loss_tracker import loss_tracker
            async with AsyncSessionLocal() as session:
@@ -328,16 +299,14 @@ async def lifespan(app: FastAPI):
        except Exception as e:
            logger.warning(f"⚠️ loss_tracker 초기화 실패 (non-critical): {e}")

-        # 테스트 데이터 및 오래된 데이터 자동 정리
        try:
            from app.models.models import StockAnalysis, TradeHistory, Portfolio
            from sqlalchemy import delete
-            from datetime import datetime, date, timedelta
-            
+            from datetime import datetime, timedelta
+
            async with AsyncSessionLocal() as session:
                _kst = pytz.timezone("Asia/Seoul")
                _today_kst = datetime.now(_kst).date()
-                # 오래된 stock_analysis 삭제 (7일 이전)
                cutoff_date = _today_kst - timedelta(days=7)
                result = await session.execute(
                    delete(StockAnalysis).where(StockAnalysis.analysis_date < cutoff_date)
@@ -345,50 +314,48 @@ async def lifespan(app: FastAPI):
                if result.rowcount > 0:
                    logger.info(f"🗑️ Deleted {result.rowcount} old stock_analysis records")

-                # PENDING/FAILED 오래된 거래 삭제 (1일 이전)
                cutoff_datetime = _today_kst - timedelta(days=1)
                result = await session.execute(
                    delete(TradeHistory).where(
                        TradeHistory.status.in_(['PENDING', 'FAILED']),
-                        TradeHistory.created_at < cutoff_datetime
+                        TradeHistory.created_at < cutoff_datetime,
                    )
                )
                if result.rowcount > 0:
                    logger.info(f"🗑️ Deleted {result.rowcount} old PENDING/FAILED trades")
-                
-                # 수량 0인 포트폴리오 삭제
+
                result = await session.execute(
                    delete(Portfolio).where(Portfolio.quantity <= 0)
                )
                if result.rowcount > 0:
                    logger.info(f"🗑️ Deleted {result.rowcount} empty portfolio records")
-                
+
                await session.commit()
                logger.info("✅ Test data cleanup completed")
        except Exception as e:
            logger.warning(f"⚠️ Test data cleanup failed (non-critical): {e}")
-            
+
    except ConnectionRefusedError as e:
        logger.error(f"❌ Database connection refused: {e}")
-        logger.error("Please check if PostgreSQL server is running and accessible")
        raise
    except Exception as e:
        logger.error(f"❌ Failed to create database tables: {e}")
        import traceback
        logger.error(traceback.format_exc())
        raise
-    
-    # 데이트레이딩 포지션 복구 (재시작 시 인메모리 유실 방지)
+
    try:
        from app.api.endpoints.day_trading import restore_positions_from_db
        await restore_positions_from_db()
    except Exception as e:
        logger.warning(f"⚠️ 데이트레이딩 포지션 복구 실패 (non-critical): {e}")

-    # 워커 시작 (스케줄러 및 API 요청 처리 전 큐 준비)
-    app.state.db_worker_task = asyncio.create_task(db_writer_worker())
+    # ── httpx 싱글톤 클라이언트 초기화 (이슈 6) ─────────────────────────
+    from app.api.endpoints.mcp import init_http_client, close_http_client
+    await init_http_client()

-    # 스케줄러 시작
+    # 워커 + 스케줄러 시작
+    app.state.db_worker_task = asyncio.create_task(db_writer_worker())
    app.state.auto_trade_task = asyncio.create_task(auto_trade_scheduler())
    app.state.batch_analysis_task = asyncio.create_task(batch_analysis_scheduler())
    app.state.day_trade_task = asyncio.create_task(day_trade_scheduler())
@@ -400,13 +367,13 @@ async def lifespan(app: FastAPI):

    yield

-    # 종료 시 정리 작업
+    # 태스크 정리
    for task_attr in (
        "auto_trade_task",
        "batch_analysis_task",
        "day_trade_task",
        "db_worker_task",
-        "kis_stream_task",   # ← 추가
+        "kis_stream_task",
    ):
        try:
            task = getattr(app.state, task_attr, None)
@@ -417,6 +384,12 @@ async def lifespan(app: FastAPI):
        except Exception as e:
            logger.warning(f"Failed to stop {task_attr} gracefully: {e}")

+    # ── httpx 클라이언트 종료 (이슈 6) ──────────────────────────────────
+    try:
+        await close_http_client()
+    except Exception as e:
+        logger.warning(f"Failed to close http_client: {e}")
+
    logger.info("Shutting down Auto Stock Trading System...")
    await engine.dispose()

@@ -429,8 +402,7 @@ app = FastAPI(
    debug=settings.DEBUG,
 )

-# CORS 설정 (반드시 라우터 등록 전에)
-# allow_origins=["*"] + allow_credentials=True 조합은 CORS 스펙 위반으로 브라우저 차단됨
+# CORS 설정
 _cors_origins = getattr(settings, "CORS_ORIGINS", None)
 if _cors_origins:
    _allowed_origins = [o.strip() for o in _cors_origins.split(",") if o.strip()]
@@ -449,33 +421,28 @@ app.add_middleware(
    allow_headers=["*"],
    expose_headers=["*"],
 )
-
 logger.info("CORS middleware configured")

-# 요청 로깅 미들웨어
+
@app.middleware("http")
 async def log_requests(request: Request, call_next):
    start_time = time.time()
    logger.info(f"🔵 Request: {request.method} {request.url}")
-    
    response = await call_next(request)
-    
    process_time = time.time() - start_time
-    logger.info(f"🟢 Response: {request.method} {request.url} - Status: {response.status_code} - Time: {process_time:.3f}s")
-    
+    logger.info(
+        f"🟢 Response: {request.method} {request.url} "
+        f"- Status: {response.status_code} - Time: {process_time:.3f}s"
+    )
    return response

-# API 라우터 등록
+
 app.include_router(api_router, prefix="/api/v1")


@app.get("/")
 async def root():
-    return {
-        "message": "Auto Stock Trading System API",
-        "version": "1.0.0",
-        "status": "running"
-    }
+    return {"message": "Auto Stock Trading System API", "version": "1.0.0", "status": "running"}


@app.get("/health")
@@ -488,22 +455,11 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
    """WebSocket 연결 엔드포인트.

    클라이언트→서버 지원 프로토콜:
-      {"type": "ping"}
-        → {"type": "pong"}
-
-      {"type": "subscribe", "stock_codes": ["005930", ...]}
-        → {"type": "ack", "action": "subscribe",
-           "stock_codes": [...], "subscriptions": [현재 전체 목록]}
-
-      {"type": "unsubscribe", "stock_codes": ["005930", ...]}
-        → {"type": "ack", "action": "unsubscribe",
-           "stock_codes": [...], "subscriptions": [현재 전체 목록]}
-
-      {"type": "set_subscriptions", "stock_codes": ["005930", ...]}
-        → {"type": "ack", "action": "set_subscriptions",
-           "stock_codes": [...], "subscriptions": [현재 전체 목록]}
-
-      그 외 메시지 → echo 응답
+      {"type": "ping"}                                       → {"type": "pong"}
+      {"type": "subscribe", "stock_codes": [...]}           → ACK
+      {"type": "unsubscribe", "stock_codes": [...]}         → ACK
+      {"type": "set_subscriptions", "stock_codes": [...]}   → ACK
+      그 외 메시지                                           → echo
    """
    await websocket_manager.connect(websocket, client_id)
    try:
@@ -512,28 +468,23 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
            try:
                msg = json.loads(data)
            except (json.JSONDecodeError, AttributeError):
-                await websocket_manager.send_personal_message(
-                    f"Echo: {data}", client_id
-                )
+                await websocket_manager.send_personal_message(f"Echo: {data}", client_id)
                continue

            msg_type = msg.get("type", "")

-            # ── ping / pong ──────────────────────────────────────────────
            if msg_type == "ping":
                await websocket_manager.send_personal_message(
                    json.dumps({"type": "pong"}), client_id
                )
                continue

-            # ── subscribe ───────────────────────────────────────────────
            if msg_type == "subscribe":
                codes = [str(c) for c in msg.get("stock_codes", [])]
                websocket_manager.subscribe(client_id, codes)
                await websocket_manager.send_personal_message(
                    json.dumps({
-                        "type": "ack",
-                        "action": "subscribe",
+                        "type": "ack", "action": "subscribe",
                        "stock_codes": codes,
                        "subscriptions": websocket_manager.get_subscriptions(client_id),
                    }),
@@ -541,14 +492,12 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
                )
                continue

-            # ── unsubscribe ──────────────────────────────────────────────
            if msg_type == "unsubscribe":
                codes = [str(c) for c in msg.get("stock_codes", [])]
                websocket_manager.unsubscribe(client_id, codes)
                await websocket_manager.send_personal_message(
                    json.dumps({
-                        "type": "ack",
-                        "action": "unsubscribe",
+                        "type": "ack", "action": "unsubscribe",
                        "stock_codes": codes,
                        "subscriptions": websocket_manager.get_subscriptions(client_id),
                    }),
@@ -556,14 +505,12 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
                )
                continue

-            # ── set_subscriptions ────────────────────────────────────────
            if msg_type == "set_subscriptions":
                codes = [str(c) for c in msg.get("stock_codes", [])]
                websocket_manager.set_subscriptions(client_id, codes)
                await websocket_manager.send_personal_message(
                    json.dumps({
-                        "type": "ack",
-                        "action": "set_subscriptions",
+                        "type": "ack", "action": "set_subscriptions",
                        "stock_codes": codes,
                        "subscriptions": websocket_manager.get_subscriptions(client_id),
                    }),
@@ -571,10 +518,7 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
                )
                continue

-            # ── 기타 메시지 → echo ───────────────────────────────────────
-            await websocket_manager.send_personal_message(
-                f"Echo: {data}", client_id
-            )
+            await websocket_manager.send_personal_message(f"Echo: {data}", client_id)

    except WebSocketDisconnect:
        websocket_manager.disconnect(client_id)
--- a/backend/app/services/ml_ensemble.py
+++ b/backend/app/services/ml_ensemble.py
@@ -9,6 +9,7 @@ ML 앙상블 모델 서비스
 앙상블 스코어: 0~100 (높을수록 매수 신호 강함)
 """

+import asyncio
 import logging
 import os
 from pathlib import Path
@@ -275,7 +276,6 @@ class MLEnsemble:
            metrics["lgbm_accuracy"] = float(lgbm_acc)
            logger.info(f"LightGBM 검증 정확도: {lgbm_acc:.4f}")

-            # 피처 중요도
            importance = self.lgbm_model.feature_importances_
            top_features = sorted(
                zip(FEATURE_COLUMNS, importance),
@@ -302,7 +302,6 @@ class MLEnsemble:
                X_seq_train, y_train,
                epochs=50, lr=0.001, batch_size=32,
            )
-            # 검증
            lstm_val_probs = self.lstm_model.predict_proba(X_seq_val)
            lstm_val_pred = np.argmax(lstm_val_probs, axis=1)
            lstm_acc = np.mean(lstm_val_pred == y_val)
@@ -342,7 +341,6 @@ class MLEnsemble:
            metrics["ensemble_accuracy"] = float(ensemble_acc)
            logger.info(f"앙상블 검증 정확도: {ensemble_acc:.4f}")

-        # 모델 저장
        self._save_models()
        self.is_trained = True
        self.training_metrics = metrics
@@ -355,7 +353,6 @@ class MLEnsemble:
        probs_list = []
        weights_list = []

-        # LightGBM
        if self.lgbm_model is not None:
            try:
                lgbm_probs = self.lgbm_model.predict_proba(X_tabular)
@@ -364,7 +361,6 @@ class MLEnsemble:
            except Exception:
                pass

-        # LSTM
        if self.lstm_model is not None and self.lstm_model.model is not None:
            try:
                lstm_probs = self.lstm_model.predict_proba(X_seq)
@@ -373,7 +369,6 @@ class MLEnsemble:
            except Exception:
                pass

-        # RF
        if self.rf_model is not None:
            try:
                rf_probs = self.rf_model.predict_proba(X_tabular)
@@ -385,7 +380,6 @@ class MLEnsemble:
        if not probs_list:
            return None

-        # 가중 평균
        total_weight = sum(weights_list)
        weighted_probs = sum(
            p * (w / total_weight) for p, w in zip(probs_list, weights_list)
@@ -393,19 +387,15 @@ class MLEnsemble:
        return weighted_probs

    def predict(self, chart_data: list) -> Dict[str, Any]:
-        """단일 종목 예측.
+        """단일 종목 예측 (동기).
+
+        이벤트 루프 내 async 핸들러에서는 predict_async() 를 사용하세요.

        Args:
            chart_data: KIS API 차트 데이터 (최신순)

        Returns:
-            {
-                "ensemble_score": int (0~100),
-                "confidence": float (0~1),
-                "signal": str ("BUY"/"HOLD"/"SELL"),
-                "model_scores": {"lgbm": ..., "lstm": ..., "rf": ...},
-                "probabilities": {"up": ..., "hold": ..., "down": ...}
-            }
+            {ensemble_score, confidence, signal, model_scores, probabilities}
        """
        if not self.is_trained:
            return {"error": "모델이 학습되지 않았습니다", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"}
@@ -414,25 +404,18 @@ class MLEnsemble:
        if tabular is None or sequence is None:
            return {"error": "피처 추출 실패 (데이터 부족)", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"}

-        # 스케일링
        tabular_scaled = self.scaler.transform(tabular.reshape(1, -1))
        sequence_batch = sequence.reshape(1, 60, len(LSTM_SEQUENCE_FEATURES))

-        # 앙상블 예측
        probs = self._ensemble_predict_proba(tabular_scaled, sequence_batch)
        if probs is None:
            return {"error": "모든 모델 예측 실패", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"}

        prob_down, prob_hold, prob_up = probs[0]
-
-        # 앙상블 스코어: UP 확률 기반 (0~100)
        ensemble_score = int(prob_up * 100)
-
-        # 신뢰도: 최고 확률과 나머지의 차이
        max_prob = max(prob_down, prob_hold, prob_up)
        confidence = float(max_prob)

-        # 신호 결정
        if prob_up > prob_down and prob_up > prob_hold and prob_up > 0.4:
            signal = "BUY"
        elif prob_down > prob_up and prob_down > prob_hold and prob_down > 0.4:
@@ -440,7 +423,6 @@ class MLEnsemble:
        else:
            signal = "HOLD"

-        # 개별 모델 스코어
        model_scores = {}
        if self.lgbm_model is not None:
            try:
@@ -473,6 +455,22 @@ class MLEnsemble:
            },
        }

+    # ─── 비동기 래퍼 (이슈 7: 이벤트 루프 블로킹 방지) ──────────────────
+    async def predict_async(self, chart_data: list) -> Dict[str, Any]:
+        """asyncio.to_thread로 CPU 추론을 워커 스레드에서 실행.
+
+        async 핸들러/스케줄러에서는 이 메서드를 사용하세요.
+        무거운 RandomForest 배치 학습에는 ProcessPoolExecutor도 고려 가능.
+        """
+        return await asyncio.to_thread(self.predict, chart_data)
+
+    async def train_async(
+        self, X_tabular: np.ndarray, X_seq: np.ndarray, y: np.ndarray
+    ) -> Dict[str, Any]:
+        """asyncio.to_thread로 CPU 학습을 워커 스레드에서 실행."""
+        return await asyncio.to_thread(self.train, X_tabular, X_seq, y)
+    # ─────────────────────────────────────────────────────────────────────
+
    def _save_models(self):
        """모델 파일 저장"""
        self._ensure_dir()
@@ -486,7 +484,6 @@ class MLEnsemble:
        if self.scaler is not None:
            joblib.dump(self.scaler, self.model_dir / "scaler.pkl")

-        # 메트릭 저장
        import json
        with open(self.model_dir / "metrics.json", "w") as f:
            json.dump(self.training_metrics, f, indent=2, ensure_ascii=False, default=str)
@@ -519,7 +516,6 @@ class MLEnsemble:
                )
                self.lstm_model.load(lstm_path)

-            # 메트릭 로드
            metrics_path = self.model_dir / "metrics.json"
            if metrics_path.exists():
                import json
--- a/backend/app/services/tensorflow_predictor.py
+++ b/backend/app/services/tensorflow_predictor.py
@@ -1,3 +1,5 @@
+import asyncio
+import functools
 import logging
 from typing import List, Dict, Any, Optional

@@ -19,6 +21,9 @@ class TensorFlowPredictor:
    - If TensorFlow or model/scaler files are missing, raises TensorFlowNotAvailable
      so callers can handle gracefully (e.g., return 503 or a friendly message).
    - Fetches recent daily closes from KIS API.
+
+    이슈 7: model.predict()는 CPU-bound 동기 호출이므로 predict_stock() 내에서
+    asyncio.to_thread()로 워커 스레드에 위임해 이벤트 루프 블로킹을 방지한다.
    """

    def __init__(self, model_path: str = "models/lstm_model.h5", scaler_path: str = "models/scaler.pkl", time_steps: int = 60):
@@ -27,7 +32,6 @@ class TensorFlowPredictor:
        self.time_steps = time_steps

        try:
-            # Local import to avoid hard dependency when not installed
            import tensorflow as tf  # noqa: F401
            from tensorflow import keras  # noqa: F401
            import joblib  # noqa: F401
@@ -35,7 +39,6 @@ class TensorFlowPredictor:
            logger.warning(f"TensorFlow or joblib not available: {e}")
            raise TensorFlowNotAvailable("TensorFlow/joblib not installed in backend environment")

-        # Load resources
        try:
            from tensorflow import keras
            import joblib
@@ -47,20 +50,16 @@ class TensorFlowPredictor:
            raise TensorFlowNotAvailable(f"Model or scaler not found: {e}")

    async def fetch_kis_closes(self, stock_code: str, days: int = 60) -> np.ndarray:
-        """Fetch recent daily closes from KIS. Returns array oldest→latest.
-        Uses korea_investment_api.get_daily_chart(stock_code, "D").
-        """
+        """Fetch recent daily closes from KIS. Returns array oldest→latest."""
        try:
            resp = await korea_investment_api.get_daily_chart(stock_code, "D")
            output = resp.get("output") or []
            if not output:
                raise RuntimeError("empty_chart")
-            # KIS usually returns newest first; reverse to chronological
            closes_desc = [float(item.get("stck_clpr", 0) or 0) for item in output]
            closes = list(reversed(closes_desc))
            arr = np.array(closes[-days:], dtype=float)
            if arr.size < days:
-                # If fewer than requested, still return what's available
                logger.info(f"{stock_code}: only {arr.size} closes available (requested {days})")
            return arr
        except Exception as e:
@@ -68,26 +67,34 @@ class TensorFlowPredictor:
            raise

    async def predict_stock(self, stock_code: str) -> Dict[str, Any]:
-        """Return prediction dict with expected_return and simple signal."""
+        """Return prediction dict with expected_return and simple signal.
+
+        model.predict()를 asyncio.to_thread()로 워커 스레드에서 실행해
+        Keras/TF의 GIL-holding 연산이 이벤트 루프를 막지 않게 한다.
+        """
        try:
            prices = await self.fetch_kis_closes(stock_code, days=self.time_steps)
            if prices.size == 0:
                raise RuntimeError("no_price_data")

-            # Normalize and shape for LSTM (1, time_steps, 1)
            scaled = self.scaler.transform(prices.reshape(-1, 1))
            if scaled.shape[0] < self.time_steps:
-                # Pad at the front if needed
                pad = np.repeat(scaled[0:1], self.time_steps - scaled.shape[0], axis=0)
                scaled = np.vstack([pad, scaled])
            X_input = scaled[-self.time_steps:].reshape(1, self.time_steps, 1)

-            # Predict
-            predicted_scaled = self.model.predict(X_input, verbose=0)
+            # ── 이슈 7: CPU-bound model.predict → 워커 스레드로 분리 ──────
+            predict_fn = functools.partial(self.model.predict, X_input, verbose=0)
+            predicted_scaled = await asyncio.to_thread(predict_fn)
+            # ─────────────────────────────────────────────────────────────
+
            predicted_price = float(self.scaler.inverse_transform(predicted_scaled)[0][0])
            current_price = float(prices[-1])

-            expected_return = ((predicted_price - current_price) / current_price * 100.0) if current_price > 0 else 0.0
+            expected_return = (
+                (predicted_price - current_price) / current_price * 100.0
+                if current_price > 0 else 0.0
+            )
            signal = "BUY" if predicted_price > current_price * 1.02 else "HOLD"

            return {
@@ -101,7 +108,6 @@ class TensorFlowPredictor:
            raise
        except Exception as e:
            logger.error(f"Prediction failed for {stock_code}: {e}")
-            # Degrade gracefully
            return {
                "stock_code": stock_code,
                "error": str(e),