diff --git a/backend/app/api/endpoints/mcp.py b/backend/app/api/endpoints/mcp.py index 4a107e8..171921d 100644 --- 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)}") diff --git a/backend/app/main.py b/backend/app/main.py index 5643d50..e49ec24 100644 --- 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) diff --git a/backend/app/services/ml_ensemble.py b/backend/app/services/ml_ensemble.py index 5dac4a1..195b1fb 100644 --- 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 diff --git a/backend/app/services/tensorflow_predictor.py b/backend/app/services/tensorflow_predictor.py index 75ca2f1..3a25acf 100644 --- 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), diff --git a/mcp-server/config.py b/mcp-server/config.py index 77eb22b..dab701a 100644 --- a/mcp-server/config.py +++ b/mcp-server/config.py @@ -45,5 +45,14 @@ class Settings: CHART_DAYS = int(os.getenv("CHART_DAYS", "100")) JSON_COMPACT = os.getenv("JSON_COMPACT", "false").lower() == "true" + # CORS (이슈 5) + # 서버-서버 통신용: 브라우저 credentials 불필요. + # Docker 환경: MCP_CORS_ORIGINS=http://backend:8000 + # 개발 환경: MCP_CORS_ORIGINS=http://localhost:8000,http://127.0.0.1:8000 + MCP_CORS_ORIGINS = os.getenv( + "MCP_CORS_ORIGINS", + "http://localhost:8000,http://127.0.0.1:8000,http://backend:8000", + ) + settings = Settings() diff --git a/mcp-server/main.py b/mcp-server/main.py index fcb5099..0ba1091 100644 --- a/mcp-server/main.py +++ b/mcp-server/main.py @@ -29,13 +29,22 @@ app = FastAPI( version="1.0.0" ) -# CORS 설정 +# CORS 설정 (이슈 5) +# 문제: allow_origins=["*"] + allow_credentials=True 조합은 CORS 스펙 위반 (브라우저 차단) +# 해결: 환경변수 MCP_CORS_ORIGINS로 origin 제한 + allow_credentials=False (서버-서버 통신 불필요) +_raw_cors = settings.MCP_CORS_ORIGINS +_allowed_origins = ( + [o.strip() for o in _raw_cors.split(",") if o.strip()] + if _raw_cors + else ["http://localhost:8000", "http://127.0.0.1:8000", "http://backend:8000"] +) +logger.info(f"MCP CORS allowed origins: {_allowed_origins}") app.add_middleware( CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + allow_origins=_allowed_origins, + allow_credentials=False, # 서버-서버 통신; 브라우저 세션 쿠키 불필요 + allow_methods=["GET", "POST"], + allow_headers=["Content-Type", "Authorization"], ) @@ -93,19 +102,15 @@ async def analyze(request: AnalysisRequest): try: analysis_type = request.analysis_type data = request.data - + if analysis_type == "RULE_VALIDATION": - # 규칙 검증 rule = data.get("rule") market_data = data.get("market_data") - if not rule or not market_data: raise HTTPException( status_code=400, detail="rule and market_data are required" ) - - # 뉴스 정보 추가 수집 stock_code = market_data.get("stock_code") news_data = [] if stock_code: @@ -115,80 +120,56 @@ async def analyze(request: AnalysisRequest): news_data = news_result["news"] except Exception as e: logger.warning(f"Failed to fetch news for validation: {e}") - result = await ai_client.validate_trading_rule(rule, market_data, news_data) - return { - "analysis_type": analysis_type, - "result": result - } - + return {"analysis_type": analysis_type, "result": result} + elif analysis_type == "MARKET_ANALYSIS": - # 시장 분석 stock_code = data.get("stock_code") market_data = data.get("market_data") news = data.get("news", []) - if not stock_code or not market_data: raise HTTPException( status_code=400, detail="stock_code and market_data are required" ) - result = await ai_client.analyze_market(stock_code, market_data, news) - return { - "analysis_type": analysis_type, - "result": result - } - + return {"analysis_type": analysis_type, "result": result} + elif analysis_type == "RISK_ASSESSMENT": - # 리스크 평가 - portfolio = data.get("portfolio") + # 이슈 4: request_mcp_risk_assessment()가 trade_plan/portfolio/signals 전달 + portfolio = data.get("portfolio") or {} trade_plan = data.get("trade_plan") - - if not portfolio or not trade_plan: + signals = data.get("signals") or {} + if not trade_plan: raise HTTPException( status_code=400, - detail="portfolio and trade_plan are required" + detail="trade_plan is required" ) - result = await ai_client.assess_risk(portfolio, trade_plan) - return { - "analysis_type": analysis_type, - "result": result - } - + return {"analysis_type": analysis_type, "result": result} + elif analysis_type == "FAIR_PRICE": - # 적정가 분석 stock_code = data.get("stock_code") market_data = data.get("market_data") financial_data = data.get("financial_data") - if not stock_code or not market_data: raise HTTPException( status_code=400, detail="stock_code and market_data are required" ) - result = await ai_client.calculate_fair_price( stock_code, market_data, financial_data ) - return { - "analysis_type": analysis_type, - "result": result - } - + return {"analysis_type": analysis_type, "result": result} + elif analysis_type == "RULE_OPTIMIZATION": - # 트레이딩 규칙 최적화/생성 stock_code = data.get("stock_code") market_data = data.get("market_data") - if not stock_code or not market_data: raise HTTPException( status_code=400, detail="stock_code and market_data are required" ) - - # 뉴스 정보 추가 수집 news_data = [] try: news_result = await mcp_tools.get_news(stock_code, limit=settings.NEWS_LIMIT) @@ -196,36 +177,26 @@ async def analyze(request: AnalysisRequest): news_data = news_result["news"] except Exception as e: logger.warning(f"Failed to fetch news for optimization: {e}") - result = await ai_client.optimize_trading_rules(stock_code, market_data, news_data) - return { - "analysis_type": analysis_type, - "result": result - } - + return {"analysis_type": analysis_type, "result": result} + elif analysis_type == "GENERATE_JOURNAL": - # 매매일지 생성 trade_info = data.get("trade_info") market_data = data.get("market_data") - if not trade_info or not market_data: raise HTTPException( status_code=400, detail="trade_info and market_data are required" ) - result = await ai_client.generate_trade_journal(trade_info, market_data) - return { - "analysis_type": analysis_type, - "result": result - } - + return {"analysis_type": analysis_type, "result": result} + else: raise HTTPException( status_code=400, detail=f"Unknown analysis type: {analysis_type}" ) - + except HTTPException: raise except Exception as e: