shuim
de7921d270
## CRITICAL 수정 (20건) - day_trading: global 선언 누락, loss_tracker 연동, ban 체크 추가 - trade_executor: BUY 체결가 order_price로 수정, remaining_pct 누적 계산 - risk_validator: get_current_price dict 파싱, MAX_POSITION_SIZE 동기화, DB 세션 누수 - risk_management/recommendations: date.today() → KST 변환 - mdd_manager: Lock 밖 executed_key 참조 수정 - correlation_filter: output → output2 키 수정 - portfolio: 해외 종목 USD/KRW 환율 적용 분리 - Sharpe Ratio 거래빈도 기반 연간화 - telegram_notifier: send_message() 공개 래퍼 추가 - position_manager: 앱 재시작 시 DB 포지션 복원 - swing exit_service: remaining_pct 조건 범위 수정 - run_backtest: original_qty 기반 remaining_pct 계산 - us_market_schedule: 2027년 미국 공휴일 추가 - Login.tsx: res.ok 검증 추가 - mean_reversion_strategy: RSI 로그 하드코딩 수정 ## HIGH 수정 (10건) - db_worker: DLQ 파일 백업 메커니즘 추가 - korea_investment_api: 만료 토큰 임의 연장 제거 - market_data: chart_cache 최대 100개 + TTL 만료 - main.py: rate_limit_store 1시간 미접근 IP 정리 - recommendations: fire-and-forget try/except 추가 - swing/scheduler: DB 세션 3개 분리 - Dashboard: wsMessages 전체 순회 - Layout: /health 엔드포인트 + 30초 폴링 - ml_ensemble: POST /train require_api_key 추가 - day_trading: record_trade_result_safe() 전환 ## 코드 정리 - 프론트엔드: isUsStock, formatTime 공통 유틸 추출 - 백엔드: dead code 제거, 중복 import 통합 - MCP: JSON 파싱 헬퍼 추출 (~60줄 감소) - 전체 ~100개 파일 한국어 주석 추가 - 테스트 13건 코드 기준 수정 (468 passed) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
544 lines
25 KiB
Python
544 lines
25 KiB
Python
"""
|
|
KIS WebSocket 실시간 시세 수신 서비스
|
|
|
|
4단계 흐름:
|
|
1단계: REST API(/oauth2/Approval) 호출 → approval_key 발급
|
|
2단계: KIS WebSocket 서버(실전/모의) 연결
|
|
3단계: H0STCNT0(체결가), H0STASP0(호가) 구독 메시지 전송
|
|
4단계: 서버 데이터 파싱
|
|
→ Redis SETEX 캐시 (tick:{code}, quote:{code})
|
|
→ websocket_manager.send_market_update() 브로드캐스트
|
|
|
|
- websocket_manager 구독 목록 변화 감지 → 동적 구독/해제
|
|
- 연결 끊김 시 자동 재연결 (approval_key 재발급 포함)
|
|
- lifespan()에서 asyncio.create_task()로 기동
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import time
|
|
import ssl
|
|
from typing import Optional
|
|
|
|
try:
|
|
import httpx as _httpx
|
|
except ImportError:
|
|
_httpx = None # type: ignore
|
|
|
|
try:
|
|
import websockets
|
|
from websockets.exceptions import ConnectionClosed, InvalidStatusCode
|
|
except ImportError:
|
|
websockets = None # type: ignore
|
|
ConnectionClosed = Exception # type: ignore
|
|
InvalidStatusCode = Exception # type: ignore
|
|
|
|
from app.core.config import settings
|
|
from app.core.json_helpers import _dumps
|
|
from app.core.redis import setex_json, TTL_TICK, TTL_QUOTE
|
|
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" # 주식 호가
|
|
|
|
# ── 타이밍 상수 ──────────────────────────────────────────────────────────────
|
|
# (.env 설정: KIS_SYNC_INTERVAL, KIS_RECONNECT_WAIT, KIS_APPROVAL_KEY_TTL)
|
|
_SYNC_INTERVAL = settings.KIS_SYNC_INTERVAL # 구독 목록 동기화 주기 (초)
|
|
_RECONNECT_WAIT = settings.KIS_RECONNECT_WAIT # 재연결 기본 대기 (초)
|
|
_RECONNECT_MAX_WAIT = 300 # 최대 재연결 대기 (5분) — 지수 백오프 상한
|
|
_APPROVAL_KEY_TTL = settings.KIS_APPROVAL_KEY_TTL # approval_key 캐시 유효시간 (초)
|
|
|
|
|
|
# ── 모듈 레벨 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 _create_ssl_context() -> ssl.SSLContext:
|
|
"""KIS WebSocket 연결용 SSL 컨텍스트 (검증 비활성화)."""
|
|
ctx = ssl.create_default_context()
|
|
# ctx.check_hostname = False
|
|
# ctx.verify_mode = ssl.CERT_NONE
|
|
return ctx
|
|
|
|
|
|
def _get_ws_url() -> str:
|
|
# [이슈4 수정] default값이 True이면 True == "virtual" → False → 실전 URL 반환 오류.
|
|
# KIS_MODE 기본값은 "virtual" 문자열로 지정해야 함.
|
|
return _KIS_WS_VIRTUAL if getattr(settings, "KIS_MODE", "virtual") == "virtual" 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
|
|
|
|
# KIS_MODE에 따라 REST API 엔드포인트 선택 (WebSocket 승인도 같은 서버)
|
|
is_real = getattr(settings, "KIS_MODE", "virtual").lower() == "real"
|
|
if is_real:
|
|
base_url = "https://openapi.koreainvestment.com:9443"
|
|
app_key = settings.KIS_REAL_APP_KEY or settings.KOREA_INVESTMENT_APP_KEY
|
|
app_secret = settings.KIS_REAL_APP_SECRET or settings.KOREA_INVESTMENT_APP_SECRET
|
|
else:
|
|
base_url = "https://openapivts.koreainvestment.com:29443"
|
|
app_key = settings.KIS_VIRTUAL_APP_KEY or settings.KOREA_INVESTMENT_APP_KEY
|
|
app_secret = settings.KIS_VIRTUAL_APP_SECRET or 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 (모드={['virtual','real'][is_real]})")
|
|
try:
|
|
async with _httpx.AsyncClient(timeout=30.0, verify=True) 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(f"✅ KIS WS approval_key 발급 완료 (유효: 23h, 길이={len(key)})")
|
|
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 _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) >= 38:
|
|
# H0STASP0 실제 포맷 (KIS 개발가이드 기준):
|
|
# [0]=종목코드, [1]=영업시간, [2]=시간구분코드
|
|
# [3~12]=매도호가1~10, [13~22]=매수호가1~10
|
|
# [23~32]=매도호가잔량1~10, [33~42]=매수호가잔량1~10
|
|
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(13), "bid2": _f(14), "bid3": _f(15), "bid4": _f(16), "bid5": _f(17),
|
|
"ask_qty1": _i(23), "ask_qty2": _i(24), "ask_qty3": _i(25),
|
|
"ask_qty4": _i(26), "ask_qty5": _i(27),
|
|
"bid_qty1": _i(33), "bid_qty2": _i(34), "bid_qty3": _i(35),
|
|
"bid_qty4": _i(36), "bid_qty5": _i(37),
|
|
}
|
|
|
|
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
|
|
|
|
|
|
# ── Redis 캐시 헬퍼 ──────────────────────────────────────────────────────────
|
|
async def _cache_realtime_data(msg_type: str, stock_code: str, data: dict) -> None:
|
|
"""실시간 데이터를 Redis에 SETEX로 캐시.
|
|
|
|
키 스키마:
|
|
tick:{code} - 최신 체결가/거래량 (TTL: TTL_TICK 초)
|
|
quote:{code} - 최신 매도/매수 호가 5단계 (TTL: TTL_QUOTE 초)
|
|
|
|
Redis 미연결 시 setex_json()이 조용히 False 반환 → 서비스 영향 없음.
|
|
"""
|
|
if msg_type == "tick":
|
|
await setex_json(f"tick:{stock_code}", TTL_TICK, data)
|
|
elif msg_type == "quote":
|
|
await setex_json(f"quote:{stock_code}", TTL_QUOTE, data)
|
|
|
|
|
|
# ── 스트림 클래스 ────────────────────────────────────────────────────────────
|
|
class KisRealtimeStream:
|
|
"""KIS WebSocket 실시간 스트림 관리 클래스.
|
|
|
|
사용 흐름:
|
|
run() 진입
|
|
→ 1단계: _fetch_approval_key() REST 호출
|
|
→ 2단계: websockets.connect() WebSocket 연결
|
|
→ 3단계: _sync_subscriptions() 구독 메시지 전송
|
|
→ 4단계: async for raw_msg in ws: 수신 → 파싱
|
|
→ Redis SETEX 캐시 (_cache_realtime_data)
|
|
→ websocket_manager 브로드캐스트
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._subscribed_codes: set[str] = set()
|
|
self._ws = None
|
|
self._running = False
|
|
self._approval_key: Optional[str] = None
|
|
self._sync_event: asyncio.Event = asyncio.Event()
|
|
|
|
def notify_subscription_change(self):
|
|
"""구독 목록 변경 시 즉시 KIS 동기화 트리거 (set_subscriptions 호출 직후 사용)"""
|
|
self._sync_event.set()
|
|
|
|
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 재발급 후 재연결.
|
|
|
|
주의: KIS WebSocket이 응답하지 않으면 일부 설정을 확인하십시오:
|
|
- .env: KIS_WS_VIRTUAL_URL / KIS_WS_REAL_URL (포트, 경로)
|
|
- KIS 모드: KIS_MODE (virtual/real)
|
|
- 재시도: 최대 5회 후 경고만 표시 (무한 루프 방지)
|
|
"""
|
|
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})")
|
|
|
|
connection_attempts = 0
|
|
max_connection_attempts = 5 # 5회 실패 후 경고만
|
|
|
|
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:
|
|
ssl_context = _create_ssl_context()
|
|
|
|
# WebSocket URL: KIS 공식 WebSocket은 경로 없이 루트(/)로 연결
|
|
# config에 이미 /websocket 등 경로가 포함되어 있으면 그대로 사용
|
|
ws_full_url = ws_url
|
|
|
|
if connection_attempts < max_connection_attempts:
|
|
logger.info(f"🚀 KIS WS 접속 시도 #{connection_attempts + 1}: {ws_full_url} (key_len={len(self._approval_key) if self._approval_key else 0})")
|
|
else:
|
|
if connection_attempts == max_connection_attempts:
|
|
logger.warning(
|
|
f"⚠️ KIS WS {max_connection_attempts}회 연결 실패. 앞으로 30초마다 시도합니다.\n"
|
|
f" 확인 사항:\n"
|
|
f" 1. .env에서 KIS_WS_VIRTUAL_URL / KIS_WS_REAL_URL 정확한가?\n"
|
|
f" 2. KIS_MODE가 'virtual' 또는 'real'로 설정되었나?\n"
|
|
f" 3. 네트워크에서 {ws_full_url}에 접근 가능한가?\n"
|
|
f" 4. KIS 서버가 WebSocket을 지원하나? (REST API만 지원할 수도 있음)\n"
|
|
)
|
|
logger.debug(f"🔄 KIS WS 재연결 시도 #{connection_attempts + 1}...")
|
|
# Allow optional override via settings, otherwise derive an Origin from the ws url
|
|
origin = getattr(settings, "KIS_WS_ORIGIN", None)
|
|
if not origin:
|
|
try:
|
|
if ws_full_url.startswith("wss://"):
|
|
origin = "https://" + ws_full_url[len("wss://"):].split("/")[0]
|
|
elif ws_full_url.startswith("ws://"):
|
|
origin = "http://" + ws_full_url[len("ws://"):].split("/")[0]
|
|
except Exception:
|
|
origin = None
|
|
|
|
# 헤더에 Origin 반드시 포함
|
|
extra_headers = {
|
|
"Origin": "https://securities.koreainvestment.com",
|
|
"User-Agent": "Mozilla/5.0"
|
|
}
|
|
if origin:
|
|
extra_headers["Origin"] = origin
|
|
logger.debug(f"📡 WebSocket extra headers: {extra_headers}")
|
|
try:
|
|
# Optional subprotocols can be set via settings.KIS_WS_SUBPROTOCOLS
|
|
subprotocols = getattr(settings, "KIS_WS_SUBPROTOCOLS", None)
|
|
if isinstance(subprotocols, str):
|
|
subprotocols = [s.strip() for s in subprotocols.split(",") if s.strip()]
|
|
if subprotocols:
|
|
logger.info(f"🧩 Attempting WebSocket subprotocols: {subprotocols}")
|
|
|
|
ws_connect_kwargs = dict(
|
|
ping_interval=30,
|
|
ping_timeout=10,
|
|
close_timeout=5,
|
|
#ssl=ssl_context,
|
|
extra_headers=extra_headers,
|
|
)
|
|
if subprotocols:
|
|
ws_connect_kwargs["subprotocols"] = subprotocols
|
|
|
|
async with websockets.connect(ws_full_url, **ws_connect_kwargs) as ws:
|
|
self._ws = ws
|
|
connection_attempts = 0 # Reset on success
|
|
logger.info("✅ KIS WebSocket 연결 완료")
|
|
|
|
# ── 3단계: 초기 구독 ──────────────────────────────────
|
|
await self._sync_subscriptions(ws)
|
|
last_sync = time.monotonic()
|
|
|
|
# ── 4단계: 데이터 수신 루프 ──────────────────────────
|
|
async for raw_msg in ws:
|
|
# 주기적 구독 동기화 (10초 주기 OR 구독 변경 즉시)
|
|
if self._sync_event.is_set() or time.monotonic() - last_sync >= _SYNC_INTERVAL:
|
|
self._sync_event.clear()
|
|
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
|
|
|
|
# 체결/호가 → Redis 캐시 + 클라이언트 브로드캐스트
|
|
if msg_type in ("tick", "quote"):
|
|
stock_code = parsed.get("stock_code", "")
|
|
if stock_code:
|
|
# 1) Redis SETEX 캐시 (실패해도 non-blocking)
|
|
await _cache_realtime_data(msg_type, stock_code, parsed)
|
|
# 2) WebSocket 클라이언트 브로드캐스트
|
|
await websocket_manager.send_market_update(stock_code, parsed)
|
|
except InvalidStatusCode as isc:
|
|
# WebSocket handshake returned a non-101 HTTP response (often 200)
|
|
connection_attempts += 1
|
|
status = getattr(isc, 'status_code', None)
|
|
logger.error(f"❌ KIS WS InvalidStatusCode: HTTP {status}")
|
|
# Try HTTP GET diagnostic (replace wss:// -> https://)
|
|
if _httpx is not None:
|
|
try:
|
|
http_url = ws_full_url.replace('wss://', 'https://')
|
|
logger.info(f"🔍 Performing HTTP GET diagnostic to {http_url}")
|
|
async with _httpx.AsyncClient(timeout=10.0, verify=True) as client:
|
|
# include Origin in diagnostic GET if available to mirror websocket handshake
|
|
diag_headers = {"User-Agent": "Mozilla/5.0"}
|
|
if origin:
|
|
diag_headers["Origin"] = origin
|
|
resp = await client.get(http_url, headers=diag_headers)
|
|
headers_snippet = dict(list(resp.headers.items())[:10])
|
|
body_snippet = (resp.text or '')[:1000]
|
|
logger.error(f"🔍 Diagnostic HTTP {resp.status_code} headers={headers_snippet} body={body_snippet}")
|
|
except Exception as diag_exc:
|
|
logger.error(f"🔍 Diagnostic HTTP GET failed: {diag_exc}", exc_info=True)
|
|
else:
|
|
logger.error("🔍 httpx not installed — cannot perform HTTP diagnostic")
|
|
# reset ws state and approval_key to attempt re-issue on next loop
|
|
self._ws = None
|
|
self._subscribed_codes.clear()
|
|
self._approval_key = None
|
|
# continue to retry after outer sleep
|
|
continue
|
|
|
|
except ConnectionClosed as e:
|
|
connection_attempts += 1
|
|
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:
|
|
connection_attempts += 1
|
|
logger.error(f"❌ KIS WS 오류 (시도 #{connection_attempts}): {type(e).__name__}: {e}")
|
|
if connection_attempts <= max_connection_attempts:
|
|
logger.debug(f" 예외 상세: ", exc_info=True)
|
|
self._ws = None
|
|
self._subscribed_codes.clear()
|
|
self._approval_key = None
|
|
|
|
if self._running:
|
|
# [FIX] 지수 백오프: 연속 실패 시 대기 시간 증가 (KIS IP 차단 방지)
|
|
backoff = min(_RECONNECT_WAIT * (2 ** min(connection_attempts - 1, 6)), _RECONNECT_MAX_WAIT)
|
|
logger.info(f"⏳ {backoff:.0f}초 후 재연결 (시도 #{connection_attempts})")
|
|
await asyncio.sleep(backoff)
|
|
|
|
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()
|