diff --git a/backend/app/core/redis.py b/backend/app/core/redis.py new file mode 100644 index 0000000..1948b01 --- /dev/null +++ b/backend/app/core/redis.py @@ -0,0 +1,127 @@ +""" +Redis 클라이언트 모듈 + +- redis.asyncio 기반 싱글턴 비동기 클라이언트 +- init_redis() / close_redis() → FastAPI lifespan에서 호출 +- setex_json() / get_json() 헬퍼 함수 제공 +- 연결 실패 시 경고 로그 후 None 반환 (서비스 무중단) + +키 스키마: + tick:{code} 실시간 체결가 (TTL: TTL_TICK) + quote:{code} 실시간 호가 (TTL: TTL_QUOTE) + recommendations:latest 배치 추천 목록 (TTL: TTL_RECOMMENDATIONS) + analysis:{code} 종목별 분석 (TTL: TTL_ANALYSIS) +""" +import json +import logging +from typing import Any, Optional + +import redis.asyncio as aioredis +from redis.asyncio import Redis + +from app.core.config import settings + +logger = logging.getLogger(__name__) + +# ── TTL 상수 ───────────────────────────────────────────────────────────────── +TTL_TICK = 10 # tick:{code} 실시간 체결가 캐시 (초) +TTL_QUOTE = 10 # quote:{code} 실시간 호가 캐시 (초) +TTL_RECOMMENDATIONS = 1800 # recommendations:latest 배치 추천 캐시 (30분) +TTL_ANALYSIS = 3600 # analysis:{code} 종목별 분석 결과 (1시간) + +# ── 싱글턴 클라이언트 ────────────────────────────────────────────────────────── +_redis_client: Optional[Redis] = None + + +async def init_redis() -> None: + """FastAPI lifespan startup에서 호출. + + REDIS_URL 은 settings.REDIS_URL 사용 (기본: redis://localhost:6379/0). + 연결 실패 시 _redis_client = None 유지 → 캐시 함수들이 조용히 skip. + """ + global _redis_client + try: + client = aioredis.from_url( + settings.REDIS_URL, + encoding="utf-8", + decode_responses=True, + ) + await client.ping() + _redis_client = client + logger.info(f"✅ Redis 연결 완료: {settings.REDIS_URL}") + except Exception as e: + logger.warning(f"⚠️ Redis 연결 실패 (캐시 비활성화): {e}") + _redis_client = None + + +async def close_redis() -> None: + """FastAPI lifespan shutdown에서 호출.""" + global _redis_client + if _redis_client: + await _redis_client.aclose() + _redis_client = None + logger.info("🔴 Redis 연결 종료") + + +def get_redis() -> Optional[Redis]: + """현재 Redis 클라이언트 반환. 연결 안 됐으면 None.""" + return _redis_client + + +# ── 헬퍼 함수 ───────────────────────────────────────────────────────────────── + +async def setex_json(key: str, ttl: int, data: Any) -> bool: + """JSON 직렬화 후 SETEX. 실패 시 False 반환 (non-blocking). + + Args: + key: Redis 키 (예: "tick:005930") + ttl: 만료 시간 (초) + data: JSON 직렬화 가능한 객체 + + Returns: + True: 저장 성공 / False: Redis 미연결 또는 직렬화·네트워크 오류 + """ + client = get_redis() + if not client: + return False + try: + await client.setex(key, ttl, json.dumps(data, ensure_ascii=False)) + return True + except Exception as e: + logger.warning(f"⚠️ Redis SETEX 실패 [{key}]: {e}") + return False + + +async def get_json(key: str) -> Optional[Any]: + """Redis GET 후 JSON 역직렬화. 없거나 실패 시 None 반환. + + Args: + key: Redis 키 (예: "tick:005930") + + Returns: + 파싱된 데이터 또는 None (캐시 미스 / Redis 미연결) + """ + client = get_redis() + if not client: + return None + try: + value = await client.get(key) + if value is None: + return None + return json.loads(value) + except Exception as e: + logger.warning(f"⚠️ Redis GET 실패 [{key}]: {e}") + return None + + +async def delete_key(key: str) -> bool: + """Redis 키 삭제. 실패 시 False 반환.""" + client = get_redis() + if not client: + return False + try: + await client.delete(key) + return True + except Exception as e: + logger.warning(f"⚠️ Redis DELETE 실패 [{key}]: {e}") + return False diff --git a/backend/app/main.py b/backend/app/main.py index a13a585..112a2c9 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -351,10 +351,14 @@ async def lifespan(app: FastAPI): except Exception as e: logger.warning(f"⚠️ 데이트레이딩 포지션 복구 실패 (non-critical): {e}") - # ── httpx 싱글톤 클라이언트 초기화 (이슈 6) ───────────────────────── + # ── httpx 싱글톤 클라이언트 초기화 ────────────────────────────────────── from app.api.endpoints.mcp import init_http_client, close_http_client await init_http_client() + # ── Redis 클라이언트 초기화 ─────────────────────────────────────────────── + from app.core.redis import init_redis, close_redis + await init_redis() + # 워커 + 스케줄러 시작 app.state.db_worker_task = asyncio.create_task(db_writer_worker()) app.state.auto_trade_task = asyncio.create_task(auto_trade_scheduler()) @@ -385,12 +389,18 @@ async def lifespan(app: FastAPI): except Exception as e: logger.warning(f"Failed to stop {task_attr} gracefully: {e}") - # ── httpx 클라이언트 종료 (이슈 6) ────────────────────────────────── + # ── httpx 클라이언트 종료 ───────────────────────────────────────────────── try: await close_http_client() except Exception as e: logger.warning(f"Failed to close http_client: {e}") + # ── Redis 클라이언트 종료 ───────────────────────────────────────────────── + try: + await close_redis() + except Exception as e: + logger.warning(f"Failed to close Redis: {e}") + logger.info("Shutting down Auto Stock Trading System...") await engine.dispose() diff --git a/backend/app/services/kis_realtime_stream.py b/backend/app/services/kis_realtime_stream.py index ef655b6..654c4b7 100644 --- a/backend/app/services/kis_realtime_stream.py +++ b/backend/app/services/kis_realtime_stream.py @@ -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: