diff --git a/backend/app/api/endpoints/market_data.py b/backend/app/api/endpoints/market_data.py index 29c975f..d44215a 100644 --- a/backend/app/api/endpoints/market_data.py +++ b/backend/app/api/endpoints/market_data.py @@ -203,6 +203,50 @@ def _generate_sample_chart(stock_code: str, period: str, limit: int = 100) -> di # 차트 데이터를 위한 간단한 인메모리 캐시 딕셔너리 chart_cache = {} + +async def get_chart_data_cached(stock_code: str, period: str, limit: int) -> list: + """캐시된 차트 데이터 반환 — REST endpoint와 WebSocket 핸들러가 공유. + + 캐시 미스 시 KIS API 호출 후 chart_cache에 저장. + realtime 보정 없음 (현재가는 WebSocket tick이 처리). + """ + if period not in ("D", "W", "M"): + period = "D" + limit = min(limit, 240) + + cache_key = f"{stock_code}__{period}" + current_time = datetime.now() + + cached = chart_cache.get(cache_key) + if cached and (current_time - cached["timestamp"]).total_seconds() < 600: + return list(cached["data"]) + + if not settings.KOREA_INVESTMENT_APP_KEY or not settings.KOREA_INVESTMENT_APP_SECRET: + return _generate_sample_chart(stock_code, period, limit).get("data", []) + + data = await korea_investment_api.get_daily_chart(stock_code, period, limit=limit) + if "output2" not in data: + return [] + + chart_data = [] + for item in data["output2"]: + try: + chart_data.append({ + "date": item.get("stck_bsop_date"), + "open": float(str(item.get("stck_oprc", "0")).replace(",", "")), + "high": float(str(item.get("stck_hgpr", "0")).replace(",", "")), + "low": float(str(item.get("stck_lwpr", "0")).replace(",", "")), + "close": float(str(item.get("stck_clpr", "0")).replace(",", "")), + "volume": int(item.get("acml_vol", 0)), + }) + except (ValueError, TypeError) as e: + logger.error(f"가격 파싱 오류: {item} - {e}") + continue + + chart_cache[cache_key] = {"timestamp": current_time, "data": chart_data} + return chart_data + + @router.get("/chart/{stock_code}") async def get_stock_chart( stock_code: str, diff --git a/backend/app/core/config.py b/backend/app/core/config.py index e754006..9232529 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -61,6 +61,13 @@ class Settings(BaseSettings): KIS_ACCOUNT_NO: str = "" KIS_BASE_URL: str = "" + # WebSocket 접속키 (수동 지정 시 우선, 비어 있으면 자동 발급) + KIS_WS_APPROVAL_KEY: str = "" + + # KIS WebSocket URL (.env에서 오버라이드 가능) + KIS_WS_REAL_URL: str = "wss://openapi.koreainvestment.com:21000/websocket/realtime" + KIS_WS_VIRTUAL_URL: str = "wss://openapiwss.koreainvestment.com:29443/websocket/realtime" + def __init__(self, **kwargs): super().__init__(**kwargs) self._apply_kis_mode() @@ -193,7 +200,7 @@ class Settings(BaseSettings): API_RETRY_DELAY: float = 1.5 # 초기 재시도 대기 시간 (초) API_RETRY_BACKOFF: float = 2.0 # 지수 백오프 배율 API_MIN_REQUEST_INTERVAL: float = 0.5 # 최소 요청 간격 (초) - 초당 2회 권장 - API_MAX_CONCURRENT_REQUESTS: int = 2 # 동시 최대 요청 수 + API_MAX_CONCURRENT_REQUESTS: int = 4 # 동시 최대 요청 수 (batch와 UI 요청 동시 처리) API_TIMEOUT: float = 30.0 # API 타임아웃 (초) # Monitoring & Alerting diff --git a/backend/app/main.py b/backend/app/main.py index e49ec24..a13a585 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -18,6 +18,7 @@ import pytz from app.core.config import settings from app.api import api_router from app.services.websocket_manager import websocket_manager +from app.core.json_helpers import _dumps from app.services.db_worker import db_writer_worker from app.core.database import engine, Base, AsyncSessionLocal from app.models import models # 모델 임포트 @@ -459,6 +460,8 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str): {"type": "subscribe", "stock_codes": [...]} → ACK {"type": "unsubscribe", "stock_codes": [...]} → ACK {"type": "set_subscriptions", "stock_codes": [...]} → ACK + {"type": "get_chart", "stock_code": "...", "period": "D", "limit": 200} + → {"type": "chart_data", ...} 그 외 메시지 → echo """ await websocket_manager.connect(websocket, client_id) @@ -516,6 +519,44 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str): }), client_id, ) + # 구독 종목 차트 캐시 사전 워밍업 (백그라운드, 논블로킹) + # 사용자가 종목 클릭 전에 캐시가 완성되도록 미리 KIS API 호출 + if codes: + async def _warmup_chart_cache(stock_codes: list): + from app.api.endpoints.market_data import get_chart_data_cached, chart_cache + from datetime import datetime + for code in stock_codes: + # 이미 유효한 캐시가 있으면 건너뜀 + cached = chart_cache.get(f"{code}__D") + if cached and (datetime.now() - cached["timestamp"]).total_seconds() < 600: + continue + try: + await get_chart_data_cached(code, "D", 100) + logger.debug(f"[warmup] {code} 차트 캐시 완료") + except Exception as e: + logger.debug(f"[warmup] {code} 캐시 실패: {e}") + asyncio.create_task(_warmup_chart_cache(codes)) + continue + + if msg_type == "get_chart": + stock_code = str(msg.get("stock_code", "")).strip() + period = msg.get("period", "D") + limit = int(msg.get("limit", 200)) + if not stock_code: + continue + try: + from app.api.endpoints.market_data import get_chart_data_cached + chart_data = await get_chart_data_cached(stock_code, period, limit) + await websocket_manager.send_personal_message( + _dumps({"type": "chart_data", "stock_code": stock_code, "period": period, "data": chart_data}), + client_id, + ) + except Exception as e: + logger.error(f"get_chart 오류 ({stock_code}): {e}") + await websocket_manager.send_personal_message( + _dumps({"type": "chart_data", "stock_code": stock_code, "period": period, "data": [], "error": str(e)}), + client_id, + ) continue await websocket_manager.send_personal_message(f"Echo: {data}", client_id) diff --git a/backend/app/services/kis_realtime_stream.py b/backend/app/services/kis_realtime_stream.py index 46b4961..ef655b6 100644 --- a/backend/app/services/kis_realtime_stream.py +++ b/backend/app/services/kis_realtime_stream.py @@ -1,10 +1,14 @@ """ KIS WebSocket 실시간 시세 수신 서비스 -- KIS(한국투자증권) 실전/모의 WebSocket에 연결 -- H0STCNT0(현재가 체결), H0STASP0(호가) 실시간 수신 -- 수신 데이터 → websocket_manager.send_market_update() 전달 -- websocket_manager의 구독 목록 변화 감지 → 동적 구독/해제 +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 @@ -13,6 +17,11 @@ 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 @@ -25,67 +34,154 @@ 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" +# ── 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" # 주식 호가 +# ── 실시간 TR_ID ───────────────────────────────────────────────────────────── +TR_TICK = "H0STCNT0" # 주식 현재가 (체결 틱) +TR_QUOTE = "H0STASP0" # 주식 호가 -# 구독 변경 감지 주기 (초) -_SYNC_INTERVAL = 10 -# 재연결 대기 (초) -_RECONNECT_WAIT = 5 +# ── 타이밍 상수 ────────────────────────────────────────────────────────────── +_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 -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) +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' : 해제 +# ── 구독 메시지 빌더 ────────────────────────────────────────────────────────── +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", + "approval_key": approval_key, + "custtype": "P", + "tr_type": tr_type, + "content-type": "utf-8", }, "body": { "input": { - "tr_id": tr_id, + "tr_id": tr_id, "tr_key": stock_code, } - } + }, }) +# ── 수신 데이터 파서 ───────────────────────────────────────────────────────── def _parse_realtime_message(raw: str) -> Optional[dict]: """KIS 실시간 메시지 파싱. + - 체결/호가 데이터: 파이프(|) 구분 텍스트 - 응답/핑: JSON - 반환: {"tr_id": ..., "stock_code": ..., ...} 또는 None + + Returns: + dict 또는 None """ - if not raw or raw.startswith("{"): - # JSON 응답 (ACK / PINGPONG 등) + if not raw: + return None + + # JSON 응답 (ACK / PINGPONG) + if raw.startswith("{"): 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|데이터건수|데이터 + # 텍스트 파이프 포맷: 암호화여부|TR_ID|데이터건수|데이터 parts = raw.split("|") if len(parts) < 4: return None @@ -94,32 +190,37 @@ def _parse_realtime_message(raw: str) -> Optional[dict]: fields = body.split("^") if tr_id == TR_TICK and len(fields) >= 10: - # H0STCNT0: [0]=종목코드, [2]=현재가, [4]=전일대비율, [7]=누적거래량, [9]=누적거래대금 + # H0STCNT0 필드 순서 (KIS 개발가이드 기준) + # [0]=종목코드, [1]=체결시간, [2]=현재가, [4]=전일대비율, + # [7]=누적거래량, [9]=누적거래대금 return { - "_type": "tick", - "tr_id": tr_id, - "stock_code": fields[0], - "price": _safe_float(fields[2]), + "_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 "", + "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 + + 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, + "_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]), + "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 @@ -137,29 +238,46 @@ def _safe_int(v: str) -> Optional[int]: return None +# ── 스트림 클래스 ──────────────────────────────────────────────────────────── class KisRealtimeStream: - """KIS WebSocket 실시간 스트림 관리 클래스.""" + """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"): - """단일 종목 구독/해제 메시지 전송.""" - key = _get_approval_key() - if not key: - logger.warning("⚠️ KIS WS approval key 없음. 실시간 구독 불가.") + """단일 종목 구독/해제 메시지 전송 (체결 + 호가).""" + if not self._approval_key: return for tr_id in (TR_TICK, TR_QUOTE): - msg = _build_subscribe_msg(key, tr_id, code, tr_type) + msg = _build_subscribe_msg(self._approval_key, tr_id, code, tr_type) await ws.send(msg) - await asyncio.sleep(0.05) # rate limit 방지 + 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 + target = websocket_manager.get_all_subscribed_codes() + to_add = target - self._subscribed_codes to_remove = self._subscribed_codes - target for code in to_add: @@ -172,16 +290,27 @@ class KisRealtimeStream: 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 WS 연결 시작: {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, @@ -192,12 +321,13 @@ class KisRealtimeStream: 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() @@ -208,41 +338,53 @@ class KisRealtimeStream: msg_type = parsed.get("_type") - # PINGPONG 응답 + # PINGPONG → pong 응답 if msg_type == "pingpong": - await ws.send(raw_msg) # pong + 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 - ) + 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 실시간 스트림 종료") + 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 -# 싱글턴 인스턴스 (lifespan에서 create_task로 사용) +# ── 싱글턴 인스턴스 ────────────────────────────────────────────────────────── kis_stream = KisRealtimeStream() diff --git a/frontend/src/hooks/useWebSocket.ts b/frontend/src/hooks/useWebSocket.ts index abb4608..fc430fd 100644 --- a/frontend/src/hooks/useWebSocket.ts +++ b/frontend/src/hooks/useWebSocket.ts @@ -6,16 +6,32 @@ interface WebSocketMessage { data: unknown } -export const useWebSocket = (clientId: string) => { - const [socket, setSocket] = useState(null) +interface UseWebSocketOptions { + /** market_update 메시지 전용 콜백 (배열 누적 방지 — 실시간 틱 전용) */ + onMarketUpdate?: (data: unknown) => void + /** chart_data 메시지 전용 콜백 (배열 누적 방지 — 차트 로드 전용) */ + onChartData?: (data: unknown) => void +} + +export const useWebSocket = (clientId: string, options?: UseWebSocketOptions) => { const [messages, setMessages] = useState([]) const [connected, setConnected] = useState(false) const reconnectTimeoutRef = useRef(null) const pingIntervalRef = useRef(null) const reconnectAttemptsRef = useRef(0) const mountedRef = useRef(true) + const socketRef = useRef(null) + // 콜백을 ref로 유지 — connect 클로저가 재생성되지 않아도 최신 콜백 참조 + const onMarketUpdateRef = useRef(options?.onMarketUpdate) + const onChartDataRef = useRef(options?.onChartData) const baseDelay = 3000 + // options 변경 시 ref 동기화 + useEffect(() => { + onMarketUpdateRef.current = options?.onMarketUpdate + onChartDataRef.current = options?.onChartData + }) + const connect = useCallback(() => { if (!mountedRef.current) return @@ -39,6 +55,19 @@ export const useWebSocket = (clientId: string) => { try { const message = JSON.parse(event.data) as WebSocketMessage if (message.type === 'pong') return + + // market_update: 콜백으로만 전달 (배열 누적 방지 — 틱마다 수신) + if (message.type === 'market_update') { + onMarketUpdateRef.current?.(message.data) + return + } + + // chart_data: 콜백으로만 전달 (배열 누적 방지 — 차트 로드 전용) + if (message.type === 'chart_data') { + onChartDataRef.current?.(message) + return + } + setMessages((prev: WebSocketMessage[]) => [...prev, message]) } catch (error) { console.error('Failed to parse WebSocket message:', error) @@ -52,7 +81,7 @@ export const useWebSocket = (clientId: string) => { ws.onclose = () => { console.log('🔌 WebSocket disconnected') setConnected(false) - setSocket(null) + socketRef.current = null if (pingIntervalRef.current) { clearInterval(pingIntervalRef.current) @@ -71,7 +100,7 @@ export const useWebSocket = (clientId: string) => { } } - setSocket(ws) + socketRef.current = ws } catch (error) { console.error('Failed to create WebSocket:', error) } @@ -89,17 +118,17 @@ export const useWebSocket = (clientId: string) => { if (pingIntervalRef.current) { clearInterval(pingIntervalRef.current) } - if (socket) { - socket.close() + if (socketRef.current) { + socketRef.current.close() } } }, [connect]) const sendMessage = useCallback((message: string) => { - if (socket && connected) { - socket.send(message) + if (socketRef.current && socketRef.current.readyState === WebSocket.OPEN) { + socketRef.current.send(message) } - }, [socket, connected]) + }, []) return { messages, connected, sendMessage } } diff --git a/frontend/src/pages/Dashboard.tsx b/frontend/src/pages/Dashboard.tsx index 449276f..b13d0f6 100644 --- a/frontend/src/pages/Dashboard.tsx +++ b/frontend/src/pages/Dashboard.tsx @@ -30,13 +30,54 @@ const checkMarketOpen = () => { }; const Dashboard = () => { - const { connected: wsConnected, messages: wsMessages } = useWebSocket('dashboard') + // 실시간 현재가 (KIS WebSocket tick → 즉시 반영) + const [livePrices, setLivePrices] = useState>({}) + // 실시간 호가 (KIS WebSocket quote H0STASP0 → 선택 종목만 수신) + const [liveQuote, setLiveQuote] = useState(null) + // 차트 데이터 (WebSocket chart_data 메시지로 로드) + const [chartData, setChartData] = useState(null) + const [chartLoading, setChartLoading] = useState(false) + // selectedStock / chartPeriod을 콜백 클로저에서 안전하게 참조하기 위한 ref + const selectedStockRef = useRef('') + const chartPeriodRef = useRef<'D' | 'W' | 'M'>('D') + + const handleMarketUpdate = useCallback((data: unknown) => { + const d = data as any + if (d?._type === 'tick' && d.stock_code && d.price != null) { + setLivePrices(prev => ({ ...prev, [d.stock_code]: d.price })) + } else if (d?._type === 'quote' && d.stock_code === selectedStockRef.current) { + setLiveQuote(d) + } + }, []) + + const handleChartData = useCallback((msg: unknown) => { + const m = msg as any + // 응답이 현재 선택 종목/기간과 일치할 때만 반영 (레이스 컨디션 방지) + if (m?.stock_code !== selectedStockRef.current || m?.period !== chartPeriodRef.current) return + const data = (m.data || []).map((item: any) => ({ + date: item.date, + open: Number(item.open), + high: Number(item.high), + low: Number(item.low), + close: Number(item.close), + volume: Number(item.volume), + })).reverse() + setChartData(data) + setChartLoading(false) + }, []) + + const { connected: wsConnected, messages: wsMessages, sendMessage } = useWebSocket('dashboard', { + onMarketUpdate: handleMarketUpdate, + onChartData: handleChartData, + }) // localStorage 초기값 사용 → portfolio 로드 전에 즉시 chart 쿼리 활성화 const [selectedStock, setSelectedStock] = useState( () => localStorage.getItem('selectedStock') || '' ) + selectedStockRef.current = selectedStock // 렌더마다 ref 동기화 (stale closure 방지) const [chartPeriod, setChartPeriod] = useState<'D' | 'W' | 'M'>('D') + chartPeriodRef.current = chartPeriod // 렌더마다 ref 동기화 (stale closure 방지) const queryClient = useQueryClient() const prevLatestTradeIdRef = useRef(null) @@ -106,30 +147,14 @@ const Dashboard = () => { refetchOnWindowFocus: false, }) - // 차트 데이터 - 30초 (캔들 갱신은 1분봉 기준이므로 10초는 과도) - const { data: chartData, isLoading: chartLoading } = useQuery({ - queryKey: ['chart-data', selectedStock, chartPeriod], - queryFn: async () => { - if (!selectedStock) return null - const response = await api.get(`/market-data/chart/${selectedStock}`, { - params: { period: chartPeriod, realtime: true, limit: 200 }, - }) - - const rawData = response.data.data || []; - - return rawData.map((item: any) => ({ - date: item.date, - open: Number(item.open), - high: Number(item.high), - low: Number(item.low), - close: Number(item.close), - volume: Number(item.volume) - })).reverse(); - }, - enabled: Boolean(selectedStock), - staleTime: 60000, - refetchInterval: isMarketOpen ? 30000 : false, - }) + // 차트 데이터 — WebSocket get_chart 요청으로 로드 (REST useQuery 대체) + // selectedStock / chartPeriod / wsConnected 변경 시 자동 재요청 + useEffect(() => { + if (!selectedStock || !wsConnected) return + setChartLoading(true) + setChartData(null) + sendMessage(JSON.stringify({ type: 'get_chart', stock_code: selectedStock, period: chartPeriod, limit: 100 })) + }, [selectedStock, chartPeriod, wsConnected, sendMessage]) // 호가창 - 10초 (실시간성 필요하지만 5초는 과도) const { data: orderbook } = useQuery({ @@ -145,7 +170,8 @@ const Dashboard = () => { } }, enabled: Boolean(selectedStock), - refetchInterval: isMarketOpen ? 10000 : false, + staleTime: Infinity, + // refetchInterval 제거 — WebSocket quote(H0STASP0)로 실시간 갱신 retry: 1, }) @@ -211,15 +237,77 @@ const Dashboard = () => { } }, [wsMessages, queryClient]) - // ✅ 현재가 - /combined에서 이미 받은 current_price 활용 (별도 API 호출 불필요) - const livePrices = useMemo>(() => { - const prices: Record = {} - ;(portfolio || []).forEach((s: any) => { - if (s.current_price) prices[s.stock_code] = Number(s.current_price) + // 포트폴리오 로드 시 livePrices 초기값 설정 (REST current_price → 실시간 tick으로 덮어씌워짐) + useEffect(() => { + if (!portfolio) return + setLivePrices(prev => { + const next = { ...prev } + ;(portfolio as any[]).forEach((s: any) => { + if (s.current_price && !next[s.stock_code]) { + next[s.stock_code] = Number(s.current_price) + } + }) + return next }) - return prices }, [portfolio]) + // KIS WebSocket 구독: 보유 종목 + 선택 종목 → 실시간 tick/quote 수신 + useEffect(() => { + if (!wsConnected) return + const codes = [...new Set([ + ...(portfolio || []).map((s: any) => s.stock_code as string), + ...(selectedStock ? [selectedStock] : []), + ])] + if (codes.length === 0) return + sendMessage(JSON.stringify({ type: 'set_subscriptions', stock_codes: codes })) + }, [wsConnected, portfolio, selectedStock, sendMessage]) + + // 선택 종목 변경 시 이전 호가 초기화 (새 quote 도착 전까지 REST fallback 사용) + useEffect(() => { + setLiveQuote(null) + }, [selectedStock]) + + // 실시간 차트: 마지막 캔들에 live price 반영 (close/high/low) + // livePrice가 selectedStock에 대해서만 변경될 때 재계산 + const livePrice = livePrices[selectedStock] + const liveChartData = useMemo(() => { + if (!chartData || chartData.length === 0 || !livePrice) return chartData ?? null + const last = chartData[chartData.length - 1] + return [ + ...chartData.slice(0, -1), + { + ...last, + close: livePrice, + high: Math.max(last.high, livePrice), + low: Math.min(last.low, livePrice), + }, + ] + }, [chartData, livePrice]) + + // 실시간 호가: WebSocket quote(H0STASP0) → REST 초기값 폴백 + const activeOrderbook = useMemo(() => { + if (!liveQuote) return orderbook ?? null + return { + current_price: livePrice ?? liveQuote.ask1 ?? orderbook?.current_price ?? 0, + stock_name: orderbook?.stock_name, + asks: [ + { price: liveQuote.ask1, volume: liveQuote.ask_qty1 ?? 0 }, + { price: liveQuote.ask2, volume: liveQuote.ask_qty2 ?? 0 }, + { price: liveQuote.ask3, volume: liveQuote.ask_qty3 ?? 0 }, + { price: liveQuote.ask4, volume: liveQuote.ask_qty4 ?? 0 }, + { price: liveQuote.ask5, volume: liveQuote.ask_qty5 ?? 0 }, + ].filter(a => a.price != null && a.price > 0), + bids: [ + { price: liveQuote.bid1, volume: liveQuote.bid_qty1 ?? 0 }, + { price: liveQuote.bid2, volume: liveQuote.bid_qty2 ?? 0 }, + { price: liveQuote.bid3, volume: liveQuote.bid_qty3 ?? 0 }, + { price: liveQuote.bid4, volume: liveQuote.bid_qty4 ?? 0 }, + { price: liveQuote.bid5, volume: liveQuote.bid_qty5 ?? 0 }, + ].filter(b => b.price != null && b.price > 0), + current_volume: 0, + } + }, [liveQuote, livePrice, orderbook]) + // useMemo로 Set 안정화 → React.memo 패널에 전달 시 레퍼런스 유지 const heldStockCodes = useMemo>( () => new Set((portfolio || []).map((s: any) => s.stock_code as string)), @@ -421,7 +509,7 @@ const Dashboard = () => {
{/* 차트 섹션 */} { - OrderbookPanel: 10초마다 독립 리렌더 (다른 패널 영향 없음) */}
- +