feat: KIS 실시간 스트림 서비스, WS 구독 프로토콜, MCP 리스크 승인 연동

backend/app/main.py

@@ -76,6 +76,56 @@ 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 캐시에서 추천 종목을 읽어 손절/익절 및 신규 매수를 실행한다.
@@ -113,8 +163,27 @@ async def auto_trade_scheduler():
                         rec = await get_rule_based_recommendations_from_db(session)
 
                         if rec and rec.get("recommendations_by_rule"):
-                            logger.info(f"📊 Found recommendations, executing auto-trade")
+                            # ── MCP RISK_ASSESSMENT 승인 체크 ──────────────────────
-                            await execute_auto_trade_background(rec, session)
+                            approve, reason = await _call_mcp_risk_assessment(rec)
+                            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({
+                                        "action": "log_mcp_deny",
+                                        "reason": reason,
+                                    })
+                                except Exception:
+                                    pass
+                            else:
+                                if reason:
+                                    logger.info(f"✅ MCP RISK_ASSESSMENT APPROVE. 참고: {reason}")
+                                logger.info(f"📊 Found recommendations, executing auto-trade")
+                                await execute_auto_trade_background(rec, session)
+                            # ────────────────────────────────────────────────────────
                         else:
                             logger.info("⏭️ Auto-trade skipped: no recommendations in DB")
                     else:
@@ -324,11 +393,21 @@ async def lifespan(app: FastAPI):
     app.state.batch_analysis_task = asyncio.create_task(batch_analysis_scheduler())
     app.state.day_trade_task = asyncio.create_task(day_trade_scheduler())
 
+    # KIS 실시간 스트림 시작
+    from app.services.kis_realtime_stream import run_kis_realtime_stream
+    app.state.kis_stream_task = asyncio.create_task(run_kis_realtime_stream())
+    logger.info("📡 KIS 실시간 스트림 태스크 시작")
+
     yield
 
     # 종료 시 정리 작업
-    # 스케줄러 중단
+    for task_attr in (
-    for task_attr in ("auto_trade_task", "batch_analysis_task", "day_trade_task", "db_worker_task"):
+        "auto_trade_task",
+        "batch_analysis_task",
+        "day_trade_task",
+        "db_worker_task",
+        "kis_stream_task",   # ← 추가
+    ):
         try:
             task = getattr(app.state, task_attr, None)
             if task:
@@ -406,24 +485,97 @@ async def health_check():
 
 @app.websocket("/ws/{client_id}")
 async def websocket_endpoint(websocket: WebSocket, client_id: str):
-    """WebSocket 연결 엔드포인트"""
+    """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 응답
+    """
     await websocket_manager.connect(websocket, client_id)
     try:
         while True:
             data = await websocket.receive_text()
-            # ping/pong 하트비트 처리
             try:
                 msg = json.loads(data)
-                if msg.get("type") == "ping":
-                    await websocket_manager.send_personal_message(
-                        json.dumps({"type": "pong"}), client_id
-                    )
-                    continue
             except (json.JSONDecodeError, AttributeError):
-                pass
+                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",
+                        "stock_codes": codes,
+                        "subscriptions": websocket_manager.get_subscriptions(client_id),
+                    }),
+                    client_id,
+                )
+                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",
+                        "stock_codes": codes,
+                        "subscriptions": websocket_manager.get_subscriptions(client_id),
+                    }),
+                    client_id,
+                )
+                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",
+                        "stock_codes": codes,
+                        "subscriptions": websocket_manager.get_subscriptions(client_id),
+                    }),
+                    client_id,
+                )
+                continue
+
+            # ── 기타 메시지 → echo ───────────────────────────────────────
             await websocket_manager.send_personal_message(
                 f"Echo: {data}", client_id
             )
+
     except WebSocketDisconnect:
         websocket_manager.disconnect(client_id)
         logger.info(f"Client {client_id} disconnected")

backend/app/services/kis_realtime_stream.py

@@ -0,0 +1,251 @@
+"""
+KIS WebSocket 실시간 시세 수신 서비스
+
+- KIS(한국투자증권) 실전/모의 WebSocket에 연결
+- H0STCNT0(현재가 체결), H0STASP0(호가) 실시간 수신
+- 수신 데이터 → websocket_manager.send_market_update() 전달
+- websocket_manager의 구독 목록 변화 감지 → 동적 구독/해제
+- lifespan()에서 asyncio.create_task()로 기동
+"""
+import asyncio
+import json
+import logging
+import time
+from typing import Optional
+
+try:
+    import websockets
+    from websockets.exceptions import ConnectionClosed
+except ImportError:
+    websockets = None  # type: ignore
+    ConnectionClosed = Exception  # type: ignore
+
+from app.core.config import settings
+from app.services.websocket_manager import websocket_manager
+
+logger = logging.getLogger(__name__)
+
+# KIS WebSocket URL (실전/모의 분기)
+_KIS_WS_REAL = "wss://openapi.koreainvestment.com:9443/websocket/realtime"
+_KIS_WS_VIRTUAL = "wss://openapiwss.koreainvestment.com:9443/websocket/realtime"
+
+# 실시간 TR_ID
+TR_TICK = "H0STCNT0"   # 주식 현재가 (체결 틱)
+TR_QUOTE = "H0STASP0"  # 주식 호가
+
+# 구독 변경 감지 주기 (초)
+_SYNC_INTERVAL = 10
+# 재연결 대기 (초)
+_RECONNECT_WAIT = 5
+
+
+def _get_ws_url() -> str:
+    return _KIS_WS_VIRTUAL if getattr(settings, "is_virtual_trading", True) else _KIS_WS_REAL
+
+
+def _get_approval_key() -> Optional[str]:
+    """settings에 저장된 KIS WebSocket approval key 반환."""
+    return getattr(settings, "KIS_WS_APPROVAL_KEY", None) or getattr(settings, "KOREA_INVESTMENT_WS_APPROVAL_KEY", None)
+
+
+def _build_subscribe_msg(approval_key: str, tr_id: str, stock_code: str, tr_type: str = "1") -> str:
+    """KIS WebSocket 구독 요청 메시지 생성.
+    tr_type='1' : 구독, tr_type='2' : 해제
+    """
+    return json.dumps({
+        "header": {
+            "approval_key": approval_key,
+            "custtype": "P",
+            "tr_type": tr_type,
+            "content-type": "utf-8",
+        },
+        "body": {
+            "input": {
+                "tr_id": tr_id,
+                "tr_key": stock_code,
+            }
+        }
+    })
+
+
+def _parse_realtime_message(raw: str) -> Optional[dict]:
+    """KIS 실시간 메시지 파싱.
+    - 체결/호가 데이터: 파이프(|) 구분 텍스트
+    - 응답/핑: JSON
+    반환: {"tr_id": ..., "stock_code": ..., ...} 또는 None
+    """
+    if not raw or raw.startswith("{"):
+        # JSON 응답 (ACK / PINGPONG 등)
+        try:
+            j = json.loads(raw)
+            # PINGPONG 처리
+            if j.get("header", {}).get("tr_id") == "PINGPONG":
+                return {"_type": "pingpong"}
+            return {"_type": "json", "data": j}
+        except Exception:
+            return None
+
+    # 텍스트 파이프 구분 포맷: 암호화여부|TR_ID|데이터건수|데이터
+    parts = raw.split("|")
+    if len(parts) < 4:
+        return None
+
+    _enc, tr_id, _cnt, body = parts[0], parts[1], parts[2], parts[3]
+    fields = body.split("^")
+
+    if tr_id == TR_TICK and len(fields) >= 10:
+        # H0STCNT0: [0]=종목코드, [2]=현재가, [4]=전일대비율, [7]=누적거래량, [9]=누적거래대금
+        return {
+            "_type": "tick",
+            "tr_id": tr_id,
+            "stock_code": fields[0],
+            "price": _safe_float(fields[2]),
+            "change_rate": _safe_float(fields[4]),
+            "volume": _safe_int(fields[7]),
+            "trade_amount": _safe_int(fields[9]),
+            "timestamp": fields[1] if len(fields) > 1 else "",
+        }
+    elif tr_id == TR_QUOTE and len(fields) >= 20:
+        # H0STASP0: [0]=종목코드, [3..12]=매도호가1~5+매수호가1~5
+        return {
+            "_type": "quote",
+            "tr_id": tr_id,
+            "stock_code": fields[0],
+            "ask1": _safe_float(fields[3]),
+            "ask2": _safe_float(fields[4]),
+            "ask3": _safe_float(fields[5]),
+            "bid1": _safe_float(fields[8]),
+            "bid2": _safe_float(fields[9]),
+            "bid3": _safe_float(fields[10]),
+            "ask_qty1": _safe_int(fields[13]),
+            "bid_qty1": _safe_int(fields[18]),
+        }
+    return None
+
+
+def _safe_float(v: str) -> Optional[float]:
+    try:
+        return float(v)
+    except (ValueError, TypeError):
+        return None
+
+
+def _safe_int(v: str) -> Optional[int]:
+    try:
+        return int(v)
+    except (ValueError, TypeError):
+        return None
+
+
+class KisRealtimeStream:
+    """KIS WebSocket 실시간 스트림 관리 클래스."""
+
+    def __init__(self):
+        self._subscribed_codes: set[str] = set()
+        self._ws = None
+        self._running = False
+
+    async def _send_subscribe(self, ws, code: str, tr_type: str = "1"):
+        """단일 종목 구독/해제 메시지 전송."""
+        key = _get_approval_key()
+        if not key:
+            logger.warning("⚠️ KIS WS approval key 없음. 실시간 구독 불가.")
+            return
+        for tr_id in (TR_TICK, TR_QUOTE):
+            msg = _build_subscribe_msg(key, tr_id, code, tr_type)
+            await ws.send(msg)
+            await asyncio.sleep(0.05)  # rate limit 방지
+
+    async def _sync_subscriptions(self, ws):
+        """websocket_manager 구독 목록과 KIS 구독 상태를 동기화."""
+        target = websocket_manager.get_all_subscribed_codes()
+        to_add = target - self._subscribed_codes
+        to_remove = self._subscribed_codes - target
+
+        for code in to_add:
+            logger.info(f"📡 KIS 구독 추가: {code}")
+            await self._send_subscribe(ws, code, "1")
+        for code in to_remove:
+            logger.info(f"🚫 KIS 구독 해제: {code}")
+            await self._send_subscribe(ws, code, "2")
+
+        self._subscribed_codes = target.copy()
+
+    async def run(self):
+        """메인 스트림 루프. 재연결 포함."""
+        if websockets is None:
+            logger.error("❌ websockets 패키지 미설치. pip install websockets")
+            return
+
+        self._running = True
+        ws_url = _get_ws_url()
+        logger.info(f"🔌 KIS WS 연결 시작: {ws_url}")
+
+        while self._running:
+            try:
+                async with websockets.connect(
+                    ws_url,
+                    ping_interval=30,
+                    ping_timeout=10,
+                    close_timeout=5,
+                ) as ws:
+                    self._ws = ws
+                    logger.info("✅ KIS WebSocket 연결 완료")
+
+                    # 초기 구독
+                    await self._sync_subscriptions(ws)
+                    last_sync = time.monotonic()
+
+                    async for raw_msg in ws:
+                        # 구독 동기화 (주기적)
+                        if time.monotonic() - last_sync >= _SYNC_INTERVAL:
+                            await self._sync_subscriptions(ws)
+                            last_sync = time.monotonic()
+
+                        parsed = _parse_realtime_message(raw_msg)
+                        if not parsed:
+                            continue
+
+                        msg_type = parsed.get("_type")
+
+                        # PINGPONG 응답
+                        if msg_type == "pingpong":
+                            await ws.send(raw_msg)  # pong
+                            continue
+
+                        # 체결/호가 → 클라 브로드캐스트
+                        if msg_type in ("tick", "quote"):
+                            stock_code = parsed.get("stock_code", "")
+                            if stock_code:
+                                await websocket_manager.send_market_update(
+                                    stock_code, parsed
+                                )
+
+            except ConnectionClosed as e:
+                logger.warning(f"⚡ KIS WS 연결 끊김: {e}. {_RECONNECT_WAIT}초 후 재연결...")
+                self._ws = None
+                self._subscribed_codes.clear()
+            except asyncio.CancelledError:
+                logger.info("⏹️ KIS 실시간 스트림 종료")
+                break
+            except Exception as e:
+                logger.error(f"❌ KIS WS 오류: {e}", exc_info=True)
+                self._ws = None
+                self._subscribed_codes.clear()
+
+            if self._running:
+                await asyncio.sleep(_RECONNECT_WAIT)
+
+        self._running = False
+
+    def stop(self):
+        self._running = False
+
+
+# 싱글턴 인스턴스 (lifespan에서 create_task로 사용)
+kis_stream = KisRealtimeStream()
+
+
+async def run_kis_realtime_stream():
+    """lifespan에서 create_task()로 호출되는 진입점."""
+    await kis_stream.run()

backend/app/services/websocket_manager.py

@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Dict, Set
 from fastapi import WebSocket
 from app.core.json_helpers import _dumps
 import logging
@@ -9,71 +9,125 @@ logger = logging.getLogger(__name__)
 class WebSocketManager:
     def __init__(self):
         self.active_connections: Dict[str, WebSocket] = {}
-    
+        # 클라이언트별 구독 종목 세트
+        self._subscriptions: Dict[str, Set[str]] = {}
+
+    # ------------------------------------------------------------------
+    # 연결 관리
+    # ------------------------------------------------------------------
     async def connect(self, websocket: WebSocket, client_id: str):
         """클라이언트 연결"""
         await websocket.accept()
         self.active_connections[client_id] = websocket
-        logger.info(f"Client {client_id} connected. Total connections: {len(self.active_connections)}")
+        self._subscriptions.setdefault(client_id, set())
-    
+        logger.info(f"Client {client_id} connected. Total: {len(self.active_connections)}")
+
     def disconnect(self, client_id: str):
         """클라이언트 연결 해제"""
-        if client_id in self.active_connections:
+        self.active_connections.pop(client_id, None)
-            del self.active_connections[client_id]
+        self._subscriptions.pop(client_id, None)
-            logger.info(f"Client {client_id} disconnected. Total connections: {len(self.active_connections)}")
+        logger.info(f"Client {client_id} disconnected. Total: {len(self.active_connections)}")
-    
+
+    # ------------------------------------------------------------------
+    # 구독 관리
+    # ------------------------------------------------------------------
+    def subscribe(self, client_id: str, stock_codes: list[str]):
+        """클라이언트에 종목 추가 구독."""
+        bucket = self._subscriptions.setdefault(client_id, set())
+        bucket.update(stock_codes)
+        logger.debug(f"[sub+] {client_id}: {stock_codes} → 현재 {sorted(bucket)}")
+
+    def unsubscribe(self, client_id: str, stock_codes: list[str]):
+        """클라이언트의 특정 종목 구독 해제."""
+        bucket = self._subscriptions.get(client_id, set())
+        bucket.difference_update(stock_codes)
+        logger.debug(f"[sub-] {client_id}: {stock_codes} → 현재 {sorted(bucket)}")
+
+    def set_subscriptions(self, client_id: str, stock_codes: list[str]):
+        """클라이언트의 구독 목록을 전체 교체."""
+        self._subscriptions[client_id] = set(stock_codes)
+        logger.debug(f"[sub=] {client_id}: 구독 목록 재설정 → {sorted(stock_codes)}")
+
+    def get_subscriptions(self, client_id: str) -> list[str]:
+        """클라이언트의 현재 구독 목록 반환."""
+        return sorted(self._subscriptions.get(client_id, set()))
+
+    def get_all_subscribed_codes(self) -> Set[str]:
+        """모든 활성 클라이언트에 걸쳐 구독 중인 종목 합집합 반환.
+        KisRealtimeStream이 이 목록을 KIS 측에 구독한다.
+        """
+        result: Set[str] = set()
+        for client_id in self.active_connections:
+            result.update(self._subscriptions.get(client_id, set()))
+        return result
+
+    # ------------------------------------------------------------------
+    # 메시지 전송
+    # ------------------------------------------------------------------
     async def send_personal_message(self, message: str, client_id: str):
-        """특정 클라이언트에게 메시지 전송"""
+        """특정 클라이언트에게 텍스트 메시지 전송"""
-        if client_id in self.active_connections:
+        ws = self.active_connections.get(client_id)
+        if ws:
             try:
-                await self.active_connections[client_id].send_text(message)
+                await ws.send_text(message)
             except Exception as e:
-                logger.error(f"Error sending message to {client_id}: {e}")
+                logger.error(f"Error sending to {client_id}: {e}")
                 self.disconnect(client_id)
-    
+
     async def broadcast(self, message: str):
         """모든 연결된 클라이언트에게 메시지 브로드캐스트"""
-        disconnected_clients = []
+        disconnected = []
-        for client_id, connection in self.active_connections.items():
+        for client_id, ws in list(self.active_connections.items()):
             try:
-                await connection.send_text(message)
+                await ws.send_text(message)
             except Exception as e:
                 logger.error(f"Error broadcasting to {client_id}: {e}")
-                disconnected_clients.append(client_id)
+                disconnected.append(client_id)
-        
+        for client_id in disconnected:
-        # 연결 실패한 클라이언트 제거
-        for client_id in disconnected_clients:
             self.disconnect(client_id)
-    
+
     async def broadcast_json(self, data: dict):
         """JSON 데이터를 모든 클라이언트에게 브로드캐스트"""
-        message = _dumps(data)
+        await self.broadcast(_dumps(data))
-        await self.broadcast(message)
+
-    
     async def send_market_update(self, stock_code: str, data: dict):
-        """시장 데이터 업데이트 전송"""
+        """시장 데이터를 해당 종목을 구독한 클라이언트에게만 전송.
-        message = {
+        구독자가 없으면 전체 브로드캐스트(기존 동작 유지).
+        """
+        message = _dumps({
             "type": "market_update",
             "stock_code": stock_code,
-            "data": data
+            "data": data,
-        }
+        })
-        await self.broadcast_json(message)
+        # 해당 종목 구독 클라이언트 필터링
-    
+        targets = [
+            cid for cid, codes in self._subscriptions.items()
+            if stock_code in codes and cid in self.active_connections
+        ]
+        if not targets:
+            # 구독자 없음 — 전체 브로드캐스트(fallback)
+            await self.broadcast(message)
+            return
+        disconnected = []
+        for cid in targets:
+            ws = self.active_connections.get(cid)
+            if not ws:
+                continue
+            try:
+                await ws.send_text(message)
+            except Exception as e:
+                logger.error(f"send_market_update error → {cid}: {e}")
+                disconnected.append(cid)
+        for cid in disconnected:
+            self.disconnect(cid)
+
     async def send_trade_notification(self, trade_data: dict):
         """거래 알림 전송"""
-        message = {
+        await self.broadcast_json({"type": "trade_notification", "data": trade_data})
-            "type": "trade_notification",
+
-            "data": trade_data
-        }
-        await self.broadcast_json(message)
-    
     async def send_mcp_status(self, status: dict):
         """MCP 상태 업데이트 전송"""
-        message = {
+        await self.broadcast_json({"type": "mcp_status", "data": status})
-            "type": "mcp_status",
-            "data": status
-        }
-        await self.broadcast_json(message)
 
 
 websocket_manager = WebSocketManager()