""" KIS WebSocket 실시간 시세 수신 서비스 4단계 흐름: 1단계: REST API(/oauth2/Approval) 호출 → approval_key 발급 2단계: KIS WebSocket 서버(실전/모의) 연결 3단계: H0STCNT0(체결가), H0STASP0(호가) 구독 메시지 전송 4단계: 서버 데이터 파싱 → websocket_manager.send_market_update() 브로드캐스트 - websocket_manager 구독 목록 변화 감지 → 동적 구독/해제 - 연결 끊김 시 자동 재연결 (approval_key 재발급 포함) - lifespan()에서 asyncio.create_task()로 기동 """ import asyncio import json import logging import time from typing import Optional try: import httpx as _httpx except ImportError: _httpx = None # type: ignore 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 (실전/모의 분기, .env KIS_WS_REAL_URL / KIS_WS_VIRTUAL_URL 오버라이드 가능) ── _KIS_WS_REAL = settings.KIS_WS_REAL_URL _KIS_WS_VIRTUAL = settings.KIS_WS_VIRTUAL_URL # ── 실시간 TR_ID ───────────────────────────────────────────────────────────── TR_TICK = "H0STCNT0" # 주식 현재가 (체결 틱) TR_QUOTE = "H0STASP0" # 주식 호가 # ── 타이밍 상수 ────────────────────────────────────────────────────────────── _SYNC_INTERVAL = 10 # 구독 목록 동기화 주기 (초) _RECONNECT_WAIT = 5 # 재연결 대기 (초) _APPROVAL_KEY_TTL = 82800 # approval_key 캐시 유효시간 (23시간, 초) # ── 모듈 레벨 approval_key 캐시 ────────────────────────────────────────────── _cached_approval_key: Optional[str] = None _cached_key_expires: float = 0.0 _approval_key_lock: Optional[asyncio.Lock] = None def _get_lock() -> asyncio.Lock: """이벤트 루프 생성 후 호출되므로 lazy init.""" global _approval_key_lock if _approval_key_lock is None: _approval_key_lock = asyncio.Lock() return _approval_key_lock def _get_ws_url() -> str: return _KIS_WS_VIRTUAL if getattr(settings, "is_virtual_trading", True) else _KIS_WS_REAL async def _fetch_approval_key() -> Optional[str]: """1단계: KIS REST /oauth2/Approval 호출로 WebSocket 접속키 발급. 우선순위: 1) settings.KIS_WS_APPROVAL_KEY 수동 설정값 2) 캐시된 키 (만료 전) 3) REST API 자동 발급 """ global _cached_approval_key, _cached_key_expires async with _get_lock(): # 1순위: 수동 설정 키 manual_key = getattr(settings, "KIS_WS_APPROVAL_KEY", "") or \ getattr(settings, "KOREA_INVESTMENT_WS_APPROVAL_KEY", "") if manual_key: if _cached_approval_key != manual_key: logger.info("🔑 KIS WS approval_key: 수동 설정값 사용") _cached_approval_key = manual_key _cached_key_expires = time.monotonic() + _APPROVAL_KEY_TTL return _cached_approval_key # 2순위: 캐시 유효 if _cached_approval_key and time.monotonic() < _cached_key_expires: return _cached_approval_key # 3순위: REST API 자동 발급 if _httpx is None: logger.error("❌ httpx 패키지 미설치. pip install httpx") return None base_url = settings.KOREA_INVESTMENT_API_BASE_URL app_key = settings.KOREA_INVESTMENT_APP_KEY app_secret = settings.KOREA_INVESTMENT_APP_SECRET if not app_key or not app_secret: logger.error("❌ KIS_APP_KEY / KIS_APP_SECRET 미설정. approval_key 발급 불가.") return None logger.info(f"🔑 KIS approval_key 자동 발급 요청: {base_url}/oauth2/Approval") try: async with _httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( f"{base_url}/oauth2/Approval", json={ "grant_type": "client_credentials", "appkey": app_key, "secretkey": app_secret, }, headers={"content-type": "application/json; charset=utf-8"}, ) if resp.status_code == 200: data = resp.json() key = data.get("approval_key") if key: _cached_approval_key = key _cached_key_expires = time.monotonic() + _APPROVAL_KEY_TTL logger.info("✅ KIS WS approval_key 발급 완료 (유효: 23h)") return _cached_approval_key logger.error(f"❌ approval_key 응답에 키 없음: {data}") else: logger.error( f"❌ approval_key 발급 실패: HTTP {resp.status_code} — {resp.text[:200]}" ) except Exception as exc: logger.error(f"❌ approval_key 발급 중 오류: {exc}", exc_info=True) return 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 Returns: dict 또는 None """ if not raw: return None # JSON 응답 (ACK / PINGPONG) if raw.startswith("{"): try: j = json.loads(raw) 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 필드 순서 (KIS 개발가이드 기준) # [0]=종목코드, [1]=체결시간, [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 "", } if tr_id == TR_QUOTE and len(fields) >= 20: # H0STASP0: [0]=종목코드, [3~7]=매도호가1~5, [8~12]=매수호가1~5 # [13~17]=매도잔량1~5, [18~22]=매수잔량1~5 def _f(i: int): return _safe_float(fields[i]) if len(fields) > i else None def _i(i: int): return _safe_int(fields[i]) if len(fields) > i else None return { "_type": "quote", "tr_id": tr_id, "stock_code": fields[0], "ask1": _f(3), "ask2": _f(4), "ask3": _f(5), "ask4": _f(6), "ask5": _f(7), "bid1": _f(8), "bid2": _f(9), "bid3": _f(10), "bid4": _f(11), "bid5": _f(12), "ask_qty1": _i(13), "ask_qty2": _i(14), "ask_qty3": _i(15), "ask_qty4": _i(16), "ask_qty5": _i(17), "bid_qty1": _i(18), "bid_qty2": _i(19), "bid_qty3": _i(20), "bid_qty4": _i(21), "bid_qty5": _i(22), } 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 실시간 스트림 관리 클래스. 사용 흐름: run() 진입 → 1단계: _fetch_approval_key() REST 호출 → 2단계: websockets.connect() WebSocket 연결 → 3단계: _sync_subscriptions() 구독 메시지 전송 → 4단계: async for raw_msg in ws: 수신 → 파싱 → 브로드캐스트 """ def __init__(self): self._subscribed_codes: set[str] = set() self._ws = None self._running = False self._approval_key: Optional[str] = None async def _ensure_approval_key(self) -> bool: """approval_key 확보. 없으면 REST API 호출.""" if not self._approval_key or time.monotonic() >= _cached_key_expires: self._approval_key = await _fetch_approval_key() if not self._approval_key: logger.warning("⚠️ approval_key 없음. 실시간 구독 불가.") return False return True async def _send_subscribe(self, ws, code: str, tr_type: str = "1"): """단일 종목 구독/해제 메시지 전송 (체결 + 호가).""" if not self._approval_key: return for tr_id in (TR_TICK, TR_QUOTE): msg = _build_subscribe_msg(self._approval_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): """메인 스트림 루프 (재연결 포함). 1단계 → 2단계 → 3단계 → 4단계 반복. 연결 끊김/오류 시 approval_key 재발급 후 재연결. """ if websockets is None: logger.error("❌ websockets 패키지 미설치. pip install websockets") return self._running = True ws_url = _get_ws_url() logger.info(f"🔌 KIS 실시간 스트림 시작 (URL: {ws_url})") while self._running: # ── 1단계: approval_key 발급 ────────────────────────────────── if not await self._ensure_approval_key(): logger.warning(f"⏳ approval_key 미확보. {_RECONNECT_WAIT}초 후 재시도...") await asyncio.sleep(_RECONNECT_WAIT) continue # ── 2단계: WebSocket 연결 ──────────────────────────────────── 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 연결 완료") # ── 3단계: 초기 구독 ────────────────────────────────── await self._sync_subscriptions(ws) last_sync = time.monotonic() # ── 4단계: 데이터 수신 루프 ────────────────────────── 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 → pong 응답 if msg_type == "pingpong": await ws.send(raw_msg) continue # ACK 로그 (구독 응답 등) if msg_type == "json": hdr = parsed.get("data", {}).get("header", {}) if hdr.get("tr_id"): logger.debug(f"[KIS ACK] tr_id={hdr['tr_id']} " f"rt_cd={parsed['data'].get('body', {}).get('rt_cd')}") 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() # 재연결 시 approval_key 재발급 self._approval_key = None 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() self._approval_key = None if self._running: await asyncio.sleep(_RECONNECT_WAIT) self._running = False logger.info("🔴 KIS 실시간 스트림 종료") def stop(self): self._running = False # ── 싱글턴 인스턴스 ────────────────────────────────────────────────────────── kis_stream = KisRealtimeStream() async def run_kis_realtime_stream(): """lifespan에서 create_task()로 호출되는 진입점.""" await kis_stream.run()