From 24f3082693ad948a010afbce309ef98d0b4ea84a Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Tue, 24 Feb 2026 18:16:20 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20KIS=20=EC=8B=A4=EC=8B=9C=EA=B0=84=20?= =?UTF-8?q?=EC=8A=A4=ED=8A=B8=EB=A6=BC=20=EC=84=9C=EB=B9=84=EC=8A=A4,=20WS?= =?UTF-8?q?=20=EA=B5=AC=EB=8F=85=20=ED=94=84=EB=A1=9C=ED=86=A0=EC=BD=9C,?= =?UTF-8?q?=20MCP=20=EB=A6=AC=EC=8A=A4=ED=81=AC=20=EC=8A=B9=EC=9D=B8=20?= =?UTF-8?q?=EC=97=B0=EB=8F=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/main.py | 176 +++++++++++++- backend/app/services/kis_realtime_stream.py | 251 ++++++++++++++++++++ backend/app/services/websocket_manager.py | 136 +++++++---- 3 files changed, 510 insertions(+), 53 deletions(-) create mode 100644 backend/app/services/kis_realtime_stream.py diff --git a/backend/app/main.py b/backend/app/main.py index 2c9ee1f..5643d50 100644 --- a/backend/app/main.py +++ b/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") - await execute_auto_trade_background(rec, session) + # ── MCP RISK_ASSESSMENT 승인 체크 ────────────────────── + 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 ("auto_trade_task", "batch_analysis_task", "day_trade_task", "db_worker_task"): + for task_attr in ( + "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") diff --git a/backend/app/services/kis_realtime_stream.py b/backend/app/services/kis_realtime_stream.py new file mode 100644 index 0000000..46b4961 --- /dev/null +++ b/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() diff --git a/backend/app/services/websocket_manager.py b/backend/app/services/websocket_manager.py index f52a55a..92673eb 100644 --- a/backend/app/services/websocket_manager.py +++ b/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: - del self.active_connections[client_id] - logger.info(f"Client {client_id} disconnected. Total connections: {len(self.active_connections)}") - + self.active_connections.pop(client_id, None) + self._subscriptions.pop(client_id, None) + 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 = [] - for client_id, connection in self.active_connections.items(): + disconnected = [] + 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) - - # 연결 실패한 클라이언트 제거 - for client_id in disconnected_clients: + disconnected.append(client_id) + for client_id in disconnected: self.disconnect(client_id) - + async def broadcast_json(self, data: dict): """JSON 데이터를 모든 클라이언트에게 브로드캐스트""" - message = _dumps(data) - await self.broadcast(message) - + await self.broadcast(_dumps(data)) + async def send_market_update(self, stock_code: str, data: dict): - """시장 데이터 업데이트 전송""" - message = { + """시장 데이터를 해당 종목을 구독한 클라이언트에게만 전송. + 구독자가 없으면 전체 브로드캐스트(기존 동작 유지). + """ + message = _dumps({ "type": "market_update", "stock_code": stock_code, - "data": data - } - await self.broadcast_json(message) - + "data": data, + }) + # 해당 종목 구독 클라이언트 필터링 + 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 = { - "type": "trade_notification", - "data": trade_data - } - await self.broadcast_json(message) - + await self.broadcast_json({"type": "trade_notification", "data": trade_data}) + async def send_mcp_status(self, status: dict): """MCP 상태 업데이트 전송""" - message = { - "type": "mcp_status", - "data": status - } - await self.broadcast_json(message) + await self.broadcast_json({"type": "mcp_status", "data": status}) websocket_manager = WebSocketManager()