""" KIS REST API Client 한국투자증권 REST API 핵심 클라이언트 (토큰 관리 + HTTP 요청) """ import httpx from typing import Dict, Any, Optional, Tuple from datetime import datetime, timedelta import logging import asyncio import time import json import os from pathlib import Path from app.core.config import settings logger = logging.getLogger(__name__) class KISClient: """한국투자증권 REST API 클라이언트 (토큰 관리 + 안전한 HTTP 요청)""" TOKEN_CACHE_FILE = Path.home() / ".cache" / "auto-stock" / "kis_token_cache.json" @classmethod def for_real_market_data(cls) -> "KISClient": """실계좌 자격증명으로 시세조회 전용 인스턴스 생성""" if not settings.KIS_REAL_APP_KEY: raise ValueError("KIS_REAL_APP_KEY가 설정되지 않음") instance = cls.__new__(cls) instance.base_url = settings.KIS_REAL_URL instance.app_key = settings.KIS_REAL_APP_KEY instance.app_secret = settings.KIS_REAL_APP_SECRET instance.account_number = settings.KIS_REAL_ACCOUNT_NO instance.TOKEN_CACHE_FILE = Path.home() / ".cache" / "auto-stock" / "kis_token_cache_real.json" instance._init_common() logger.info(f"🔧 KIS 실계좌 시세조회 API: {instance.base_url}") return instance def __init__(self): self.base_url = settings.KOREA_INVESTMENT_API_BASE_URL self.app_key = settings.KOREA_INVESTMENT_APP_KEY self.app_secret = settings.KOREA_INVESTMENT_APP_SECRET self.account_number = settings.KOREA_INVESTMENT_ACCOUNT_NUMBER self._init_common() logger.info(f"🔧 KIS API 초기화: mode={settings.KIS_MODE}, {self.base_url}") def _init_common(self): """공통 초기화 로직""" self.access_token: Optional[str] = None self.token_expires_at: Optional[datetime] = None self.last_request_time: Optional[float] = None # 설정 기반 쓰로틀링 self.min_request_interval = settings.API_MIN_REQUEST_INTERVAL self.max_retries = settings.API_MAX_RETRIES self.retry_delay = settings.API_RETRY_DELAY self.retry_backoff = settings.API_RETRY_BACKOFF self.timeout = settings.API_TIMEOUT # 동시 요청 제한 self.semaphore = asyncio.Semaphore(settings.API_MAX_CONCURRENT_REQUESTS) self._token_lock = asyncio.Lock() # 캐시된 토큰 로드 self._load_token_from_cache() def _load_token_from_cache(self): """파일에서 토큰 로드""" try: cache_file = Path(self.TOKEN_CACHE_FILE) if cache_file.exists(): with open(cache_file, 'r') as f: cache = json.load(f) self.access_token = cache.get('access_token') expires_str = cache.get('expires_at') if expires_str: self.token_expires_at = datetime.fromisoformat(expires_str) if self.access_token and datetime.now() < self.token_expires_at: remaining = (self.token_expires_at - datetime.now()).total_seconds() / 3600 logger.info(f"💾 캐시 토큰 로드 (남은 시간: {remaining:.1f}시간)") except Exception as e: logger.warning(f"⚠️ 토큰 캐시 로드 실패: {e}") async def _save_token_to_cache(self): """토큰을 파일에 비동기 저장""" try: cache_file = Path(self.TOKEN_CACHE_FILE) cache_file.parent.mkdir(parents=True, exist_ok=True) cache = { 'access_token': self.access_token, 'expires_at': self.token_expires_at.isoformat() if self.token_expires_at else None, } await asyncio.to_thread(self._write_token_file, cache_file, cache) except Exception as e: logger.warning(f"⚠️ 토큰 캐시 저장 실패: {e}") def _write_token_file(self, cache_file: Path, cache: dict): """동기 파일 쓰기""" with open(cache_file, 'w') as f: json.dump(cache, f) os.chmod(cache_file, 0o600) async def get_access_token(self) -> str: """접근 토큰 발급 (Lock으로 중복 방지)""" if self.access_token and self.token_expires_at and datetime.now() < self.token_expires_at: return self.access_token async with self._token_lock: if self.access_token and self.token_expires_at and datetime.now() < self.token_expires_at: return self.access_token logger.info("🔑 새 토큰 발급 요청") async with httpx.AsyncClient() as client: response = await client.post( f"{self.base_url}/oauth2/tokenP", json={ "grant_type": "client_credentials", "appkey": self.app_key, "appsecret": self.app_secret }, timeout=30.0 ) if response.status_code == 200: data = response.json() self.access_token = data["access_token"] self.token_expires_at = datetime.now() + timedelta(hours=23) await self._save_token_to_cache() logger.info(f"✅ 토큰 발급 완료") return self.access_token else: raise Exception(f"토큰 발급 실패: {response.status_code} - {response.text}") async def request( self, method: str, endpoint: str, tr_id: str, extra_headers: Optional[Dict[str, str]] = None, **kwargs ) -> Tuple[Dict[str, Any], Optional[Any]]: """API 요청 (재시도 + 쓰로틀링)""" async with self.semaphore: return await self._request_with_retry(method, endpoint, tr_id, extra_headers, **kwargs) async def _request_with_retry( self, method: str, endpoint: str, tr_id: str, extra_headers: Optional[Dict[str, str]], **kwargs ) -> Tuple[Dict[str, Any], Optional[Any]]: """재시도 로직 포함 실제 요청""" start_time = time.time() for attempt in range(self.max_retries + 1): try: # 쓰로틀링 if self.last_request_time: elapsed = time.time() - self.last_request_time if elapsed < self.min_request_interval: await asyncio.sleep(self.min_request_interval - elapsed) self.last_request_time = time.time() token = await self.get_access_token() headers = { "content-type": "application/json; charset=utf-8", "authorization": f"Bearer {token}", "appkey": self.app_key, "appsecret": self.app_secret, "tr_id": tr_id, } if extra_headers: headers.update(extra_headers) timeout = httpx.Timeout(self.timeout, connect=10.0) async with httpx.AsyncClient(timeout=timeout) as client: response = await client.request( method, f"{self.base_url}{endpoint}", headers=headers, **kwargs ) if response.status_code == 200: return response.json(), response.headers # 재시도 대상 (429, 5xx) elif response.status_code in [429, 500, 502, 503, 504]: if attempt < self.max_retries: delay = self.retry_delay * (self.retry_backoff ** attempt) logger.warning(f"⏳ 재시도 {attempt + 1}/{self.max_retries}: {delay:.1f}초 대기") await asyncio.sleep(delay) continue else: raise Exception(f"최대 재시도 초과: {response.status_code}") # 4xx - 재시도 불필요 elif 400 <= response.status_code < 500: try: return response.json(), response.headers except Exception: return {"rt_cd": str(response.status_code), "msg1": response.text}, None else: raise Exception(f"API 실패: {response.status_code} - {response.text}") except httpx.TimeoutException: if attempt < self.max_retries: delay = self.retry_delay * (self.retry_backoff ** attempt) logger.warning(f"⏳ 타임아웃 재시도 {attempt + 1}/{self.max_retries}") await asyncio.sleep(delay) continue else: raise raise Exception("API 요청 실패") # 싱글톤 인스턴스 kis_client = KISClient() try: real_market_client = KISClient.for_real_market_data() except Exception: real_market_client = None