252 lines
8.4 KiB
Python
252 lines
8.4 KiB
Python
"""
|
|
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()
|