feat: Redis 싱글턴 클라이언트 추가 + kis_realtime_stream SETEX 연동
- backend/app/core/redis.py 신규 생성
- TTL 상수: TTL_TICK/QUOTE=10s, RECOMMENDATIONS=1800s, ANALYSIS=3600s
- init_redis() / close_redis() → FastAPI lifespan 훅
- setex_json() / get_json() / delete_key() 헬퍼
- Redis 미연결 시 경고 로그 후 non-blocking 계속 동작
- kis_realtime_stream.py: tick/quote 수신 즉시 Redis SETEX 캐시
- _cache_realtime_data() 모듈 함수 추가
- tick:{code} / quote:{code} 키 저장 (API 없이 현재가 조회 가능)
- main.py lifespan: init_redis() / close_redis() 호출 추가
This commit is contained in:
@@ -5,7 +5,9 @@ KIS WebSocket 실시간 시세 수신 서비스
|
||||
1단계: REST API(/oauth2/Approval) 호출 → approval_key 발급
|
||||
2단계: KIS WebSocket 서버(실전/모의) 연결
|
||||
3단계: H0STCNT0(체결가), H0STASP0(호가) 구독 메시지 전송
|
||||
4단계: 서버 데이터 파싱 → websocket_manager.send_market_update() 브로드캐스트
|
||||
4단계: 서버 데이터 파싱
|
||||
→ Redis SETEX 캐시 (tick:{code}, quote:{code})
|
||||
→ websocket_manager.send_market_update() 브로드캐스트
|
||||
|
||||
- websocket_manager 구독 목록 변화 감지 → 동적 구독/해제
|
||||
- 연결 끊김 시 자동 재연결 (approval_key 재발급 포함)
|
||||
@@ -30,6 +32,7 @@ except ImportError:
|
||||
ConnectionClosed = Exception # type: ignore
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.redis import setex_json, TTL_TICK, TTL_QUOTE
|
||||
from app.services.websocket_manager import websocket_manager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -238,6 +241,22 @@ def _safe_int(v: str) -> Optional[int]:
|
||||
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 실시간 스트림 관리 클래스.
|
||||
@@ -247,7 +266,9 @@ class KisRealtimeStream:
|
||||
→ 1단계: _fetch_approval_key() REST 호출
|
||||
→ 2단계: websockets.connect() WebSocket 연결
|
||||
→ 3단계: _sync_subscriptions() 구독 메시지 전송
|
||||
→ 4단계: async for raw_msg in ws: 수신 → 파싱 → 브로드캐스트
|
||||
→ 4단계: async for raw_msg in ws: 수신 → 파싱
|
||||
→ Redis SETEX 캐시 (_cache_realtime_data)
|
||||
→ websocket_manager 브로드캐스트
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
@@ -351,10 +372,13 @@ class KisRealtimeStream:
|
||||
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 ConnectionClosed as e:
|
||||
|
||||
Reference in New Issue
Block a user