From b1eab5305a06e33449dda85dd9c84c1bdda1364a Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 10:58:28 +0900 Subject: [PATCH] refactor(phase3): Add KIS WebSocket client implementation --- backend/app/realtime/websocket_client.py | 244 +++++++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 backend/app/realtime/websocket_client.py diff --git a/backend/app/realtime/websocket_client.py b/backend/app/realtime/websocket_client.py new file mode 100644 index 0000000..e9011eb --- /dev/null +++ b/backend/app/realtime/websocket_client.py @@ -0,0 +1,244 @@ +""" +KIS WebSocket Client + +KIS API WebSocket 연결 및 실시간 데이터 수신 +""" +import json +import logging +import asyncio +from typing import Dict, Any, Optional, Set, Callable +from datetime import datetime +import pytz + +try: + import websockets + from websockets.client import WebSocketClientProtocol +except ImportError: + websockets = None + WebSocketClientProtocol = None + +from app.core.config import settings +from app.realtime.event_handler import event_handler, EventType + +logger = logging.getLogger(__name__) + + +class KISWebSocketClient: + """ + KIS API WebSocket 클라이언트 + + 실시간 호가/체결 데이터 수신 + """ + + def __init__(self, app_key: str, app_secret: str): + if websockets is None: + raise ImportError("websockets 패키지가 설치되지 않았습니다: pip install websockets") + + self.app_key = app_key + self.app_secret = app_secret + self.ws: Optional[WebSocketClientProtocol] = None + self.subscriptions: Set[str] = set() # 구독 중인 종목코드 + self.is_connected = False + self.kst = pytz.timezone("Asia/Seoul") + + # WebSocket URL (모의투자 vs 실계좌) + if "vts" in settings.KOREA_INVESTMENT_API_BASE_URL.lower(): + self.ws_url = "ws://ops.koreainvestment.com:31000" # 모의투자 + else: + self.ws_url = "ws://ops.koreainvestment.com:21000" # 실계좌 + + logger.info(f"🔌 WebSocket 클라이언트 초기화: {self.ws_url}") + + async def connect(self): + """ +WebSocket 연결""" + try: + logger.info(f"🔌 WebSocket 연결 시도: {self.ws_url}") + self.ws = await websockets.connect(self.ws_url) + self.is_connected = True + logger.info("✅ WebSocket 연결 성공") + + await event_handler.emit(EventType.CONNECTED, {"timestamp": datetime.now(self.kst)}) + + except Exception as e: + logger.error(f"❌ WebSocket 연결 실패: {e}", exc_info=True) + self.is_connected = False + raise + + async def disconnect(self): + """ +WebSocket 연결 해제""" + if self.ws: + await self.ws.close() + self.is_connected = False + logger.info("🔌 WebSocket 연결 해제") + + await event_handler.emit(EventType.DISCONNECTED, {"timestamp": datetime.now(self.kst)}) + + async def subscribe_price(self, stock_code: str): + """ + 주식 현재가 구독 + + Args: + stock_code: 종목코드 (6자리) + """ + if not self.is_connected: + raise ConnectionError("WebSocket이 연결되지 않았습니다") + + # KIS WebSocket 프로토콜 + subscribe_msg = { + "header": { + "approval_key": self.app_key, + "custtype": "P", + "tr_type": "1", # 등록 + "content-type": "utf-8" + }, + "body": { + "input": { + "tr_id": "H0STCNT0", # 현재가 호가 + "tr_key": stock_code + } + } + } + + await self.ws.send(json.dumps(subscribe_msg)) + self.subscriptions.add(stock_code) + logger.info(f"📡 구독 등록: {stock_code}") + + async def unsubscribe_price(self, stock_code: str): + """주식 현재가 구독 해제""" + if not self.is_connected: + return + + unsubscribe_msg = { + "header": { + "approval_key": self.app_key, + "custtype": "P", + "tr_type": "2", # 해제 + "content-type": "utf-8" + }, + "body": { + "input": { + "tr_id": "H0STCNT0", + "tr_key": stock_code + } + } + } + + await self.ws.send(json.dumps(unsubscribe_msg)) + self.subscriptions.discard(stock_code) + logger.info(f"❌ 구독 해제: {stock_code}") + + async def listen(self): + """ + 메시지 수신 루프 (무한 대기) + + 수신한 데이터를 파싱하여 이벤트 발생 + """ + if not self.is_connected: + raise ConnectionError("WebSocket이 연결되지 않았습니다") + + logger.info("👂 WebSocket 메시지 수신 시작") + + try: + async for message in self.ws: + try: + await self._handle_message(message) + except Exception as e: + logger.error(f"❌ 메시지 처리 오류: {e}", exc_info=True) + await event_handler.emit(EventType.ERROR, {"error": str(e)}) + + except websockets.exceptions.ConnectionClosed: + logger.warning("⚠️ WebSocket 연결 끔김") + self.is_connected = False + await event_handler.emit(EventType.DISCONNECTED, {"reason": "connection_closed"}) + + async def _handle_message(self, message: str): + """ + 수신한 메시지 파싱 및 이벤트 발생 + + Args: + message: WebSocket 메시지 (JSON 문자열) + """ + try: + data = json.loads(message) + except json.JSONDecodeError: + # 텍스트 형식일 수 있음 (KIS API 특성) + logger.debug(f"텍스트 메시지 수신: {message[:100]}...") + return + + # TR_ID 확인 + tr_id = data.get("header", {}).get("tr_id") + + if tr_id == "H0STCNT0": # 현재가 호가 + await self._handle_price_update(data) + + elif tr_id == "H0STCNI0": # 체결 + await self._handle_execution(data) + + else: + logger.debug(f"알 수 없는 TR_ID: {tr_id}") + + async def _handle_price_update(self, data: Dict[str, Any]): + """현재가 업데이트 처리""" + body = data.get("body", {}) + output = body.get("output", {}) + + event_data = { + "stock_code": output.get("MKSC_SHRN_ISCD", ""), + "stock_name": output.get("HTS_KOR_ISNM", ""), + "current_price": int(output.get("STCK_PRPR", 0)), + "change_rate": float(output.get("PRDY_CTRT", 0)), + "volume": int(output.get("ACML_VOL", 0)), + "timestamp": datetime.now(self.kst), + } + + await event_handler.emit(EventType.PRICE_UPDATE, event_data) + + async def _handle_execution(self, data: Dict[str, Any]): + """체결 데이터 처리""" + body = data.get("body", {}) + output = body.get("output", {}) + + event_data = { + "order_id": output.get("ODNO", ""), + "stock_code": output.get("PDNO", ""), + "executed_price": int(output.get("CNTR_UNPR", 0)), + "executed_quantity": int(output.get("CNTR_QTY", 0)), + "timestamp": datetime.now(self.kst), + } + + await event_handler.emit(EventType.EXECUTION, event_data) + + def get_subscriptions(self) -> Set[str]: + """현재 구독 중인 종목 목록""" + return self.subscriptions.copy() + + async def run(self): + """ + WebSocket 클라이언트 실행 (연결 + 수신) + + 자동 재연결 지원 + """ + while True: + try: + await self.connect() + await self.listen() + + except Exception as e: + logger.error(f"❌ WebSocket 오류: {e}") + await event_handler.emit(EventType.ERROR, {"error": str(e)}) + + # 5초 후 재연결 시도 + logger.info("🔄 5초 후 재연결 시도...") + await asyncio.sleep(5) + + +# 클라이언트 인스턴스 생성 헬퍼 +def create_websocket_client() -> KISWebSocketClient: + """ +KIS WebSocket 클라이언트 인스턴스 생성""" + return KISWebSocketClient( + app_key=settings.KOREA_INVESTMENT_APP_KEY, + app_secret=settings.KOREA_INVESTMENT_APP_SECRET + )