""" 한국투자증권(KIS) REST API 클라이언트 - OAuth2 토큰 자동 발급 및 파일 캐시 (~/.cache/auto-stock/) - 국내/해외 주식 현재가, 일봉/분봉 차트 조회 - 매수/매도 주문 및 체결 확인 - CircuitBreaker: 연속 N회 실패 시 일정 시간 요청 차단 - APICallMonitor: 호출 성공률·지연시간 통계 - KIS_MODE(virtual/real)에 따라 엔드포인트 자동 전환 """ import httpx from typing import Dict, Any, Optional, Deque from datetime import datetime, timedelta import logging import asyncio import time import json import os from pathlib import Path from collections import deque from app.core.config import settings logger = logging.getLogger(__name__) # ── 차트 캐시 TTL 상수 ──────────────────────────────────────────────────────── TTL_DAILY_CHART = 300 # 일봉 캐시 5분 (장중 업데이트 주기 고려) TTL_MINUTE_CHART = 60 # 분봉 캐시 1분 (실시간성 유지) TTL_MINUTE_FULLDAY = 60 # 당일 전체 분봉 캐시 1분 TTL_OVERSEAS_CHART = 300 # 해외 일봉 캐시 5분 class CircuitBreaker: """KIS API Circuit Breaker — 연속 N회 실패 시 일정 시간 요청 차단. States: CLOSED : 정상 운영 (요청 허용) OPEN : 차단 (recovery_timeout 경과 전까지 요청 거부) HALF_OPEN: 복구 시도 (1개 요청 허용 → 성공 시 CLOSED, 실패 시 OPEN) """ def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 600): self.failure_threshold = failure_threshold # 연속 실패 허용 횟수 self.recovery_timeout = recovery_timeout # 차단 후 복구 대기 시간 (초) self.failure_count = 0 self.last_failure_time: Optional[float] = None self.state = "CLOSED" def can_request(self) -> bool: """요청 가능 여부 반환.""" if self.state == "CLOSED": return True if self.state == "OPEN": if self.last_failure_time and time.time() - self.last_failure_time >= self.recovery_timeout: self.state = "HALF_OPEN" logger.info("⚡ Circuit breaker HALF_OPEN: 복구 시도") return True return False return True # HALF_OPEN def record_success(self) -> None: """성공 기록 — 차단 해제.""" if self.state != "CLOSED": logger.info("✅ Circuit breaker CLOSED: 복구 완료") self.failure_count = 0 self.state = "CLOSED" def record_failure(self) -> None: """실패 기록 — 임계치 초과 시 차단.""" self.failure_count += 1 self.last_failure_time = time.time() if self.state == "HALF_OPEN" or self.failure_count >= self.failure_threshold: self.state = "OPEN" logger.warning( f"⚡ Circuit breaker OPEN: {self.failure_count}회 연속 실패. " f"{self.recovery_timeout}초 후 재시도" ) class APICallMonitor: """API 호출 모니터링 및 통계""" def __init__(self, window_size: int = 100): self.window_size = window_size self.call_history: Deque[Dict[str, Any]] = deque(maxlen=window_size) self.total_calls = 0 self.total_errors = 0 self.total_retries = 0 def record_call(self, success: bool, duration: float, retries: int = 0): """API 호출 기록""" self.total_calls += 1 if not success: self.total_errors += 1 self.total_retries += retries self.call_history.append({ "timestamp": time.time(), "success": success, "duration": duration, "retries": retries }) def get_error_rate(self) -> float: """최근 윈도우 내 오류율 계산""" if not self.call_history: return 0.0 errors = sum(1 for call in self.call_history if not call["success"]) return errors / len(self.call_history) def get_avg_duration(self) -> float: """평균 응답 시간""" if not self.call_history: return 0.0 return sum(call["duration"] for call in self.call_history) / len(self.call_history) def get_stats(self) -> Dict[str, Any]: """통계 정보 반환""" return { "total_calls": self.total_calls, "total_errors": self.total_errors, "total_retries": self.total_retries, "error_rate": self.get_error_rate(), "avg_duration": self.get_avg_duration(), "recent_calls": len(self.call_history) } def should_alert(self, threshold: float = 0.3) -> bool: """오류율 임계값 초과 여부""" return self.get_error_rate() > threshold # ── Redis 캐시 헬퍼 (redis.py 의존성 없이 안전하게 호출) ────────────────────── async def _cache_get(key: str) -> Optional[Any]: """Redis 캐시에서 값 조회. Redis 미연결 시 None 반환 (non-blocking).""" try: from app.core.redis import get_json return await get_json(key) except Exception: return None async def _cache_set(key: str, ttl: int, data: Any) -> None: """Redis 캐시에 값 저장. 실패 시 조용히 무시 (non-blocking).""" try: from app.core.redis import setex_json await setex_json(key, ttl, data) except Exception: pass class KoreaInvestmentAPI: """한국투자증권 API 클라이언트 (안전한 재시도 및 쓰로틀링 적용)""" # 토큰 캐시 파일 경로 — 사용자 홈 디렉토리 하위에 저장 (/tmp는 보안 취약) TOKEN_CACHE_FILE = Path.home() / ".cache" / "auto-stock" / "kis_token_cache.json" @classmethod def for_real_market_data(cls) -> "KoreaInvestmentAPI": """실계좌 자격증명으로 시세조회 전용 인스턴스 생성. 모의투자 모드에서도 실계좌 API로 시세를 조회할 때 사용한다. (모의투자 서버는 KOSDAQ 거래량 순위 등 일부 조회 API를 지원하지 않음) 주문 API는 절대 호출하지 말 것. """ 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.access_token = None instance.token_expires_at = None instance.last_request_time = None instance.TOKEN_CACHE_FILE = Path.home() / ".cache" / "auto-stock" / "kis_token_cache_real.json" instance.min_request_interval = settings.API_MIN_REQUEST_INTERVAL instance.max_retries = settings.API_MAX_RETRIES instance.retry_delay = settings.API_RETRY_DELAY instance.retry_backoff = settings.API_RETRY_BACKOFF instance.timeout = settings.API_TIMEOUT instance.semaphore = asyncio.Semaphore(settings.API_MAX_CONCURRENT_REQUESTS) instance._token_lock = asyncio.Lock() instance.monitor = APICallMonitor(window_size=settings.API_MONITORING_WINDOW) instance.circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=600) logger.info( "🔧 KIS 실계좌 시세조회 API 초기화: base_url=%s, app_key=%s...", instance.base_url, instance.app_key[:6] ) instance._load_token_from_cache() 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.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 # 기본 0.5초 (초당 2회) self.max_retries = settings.API_MAX_RETRIES # 기본 3회 self.retry_delay = settings.API_RETRY_DELAY # 기본 1.5초 self.retry_backoff = settings.API_RETRY_BACKOFF # 기본 2.0배 self.timeout = settings.API_TIMEOUT # 기본 30초 # 동시 요청 제한 (세마포어) self.semaphore = asyncio.Semaphore(settings.API_MAX_CONCURRENT_REQUESTS) # 토큰 발급 Lock (중복 발급 방지) self._token_lock = asyncio.Lock() # 모니터링 self.monitor = APICallMonitor(window_size=settings.API_MONITORING_WINDOW) # Circuit Breaker (5회 연속 실패 → 10분 차단) self.circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=600) # 시작 시 설정 로깅 (키 마스킹) logger.info(f"🔧 KIS API 초기화: mode={settings.KIS_MODE}, " f"base_url={self.base_url}, " f"app_key={self.app_key[:6]}..., " f"account={self.account_number[:4]}****{self.account_number[-2:]}") # 캐시된 토큰 로드 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}시간)") else: logger.info("💾 캐시 파일의 토큰이 만료됨") self.access_token = None self.token_expires_at = None except Exception as e: logger.warning(f"⚠️ 토큰 캐시 파일 로드 실패: {e}") def _write_token_file(self): """토큰을 파일에 동기 저장 (asyncio.to_thread 내부에서 실행)""" 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, } with open(cache_file, 'w') as f: json.dump(cache, f) # 소유자만 읽기/쓰기 가능하도록 퍼미션 설정 (보안) os.chmod(cache_file, 0o600) logger.debug("💾 토큰 캐시 파일 저장 완료") async def _save_token_to_cache(self): """토큰을 파일에 비동기 저장 (이벤트 루프 블로킹 방지)""" try: await asyncio.to_thread(self._write_token_file) except Exception as e: logger.warning(f"⚠️ 토큰 캐시 파일 저장 실패: {e}") async def _get_access_token(self) -> str: """접근 토큰 발급 (Lock으로 중복 발급 방지)""" # 캐시된 토큰이 유효하면 재사용 (23시간 유효) if self.access_token and self.token_expires_at and datetime.now() < self.token_expires_at: remaining = (self.token_expires_at - datetime.now()).total_seconds() / 3600 logger.info(f"✅ 캐시된 토큰 재사용 (남은 시간: {remaining:.1f}시간)") return self.access_token # 🔒 Lock 획득 - 동시 요청 시 하나만 토큰 발급 async with self._token_lock: # Lock 대기 중 다른 요청이 토큰 발급했을 수 있으므로 재확인 if self.access_token and self.token_expires_at and datetime.now() < self.token_expires_at: remaining = (self.token_expires_at - datetime.now()).total_seconds() / 3600 logger.info(f"✅ 대기 중 토큰 발급됨, 재사용 (남은 시간: {remaining:.1f}시간)") return self.access_token # 토큰 발급 시도 전에 마지막 시도 시간 확인 (1분 제한 대응) if hasattr(self, '_last_token_request_time'): elapsed = (datetime.now() - self._last_token_request_time).total_seconds() if elapsed < 70: # 70초 대기 (여유있게) # 기존 토큰이 있으면 계속 사용 if self.access_token: logger.warning(f"⏳ 토큰 발급 1분 제한 ({elapsed:.0f}초 경과) - 기존 토큰 재사용") # 만료 시간 연장 (추가 1시간) self.token_expires_at = datetime.now() + timedelta(hours=1) return self.access_token wait_time = 70 - elapsed logger.warning(f"⏳ 토큰 발급 1분 제한: {wait_time:.1f}초 대기 중...") await asyncio.sleep(wait_time + 1) logger.info("🔑 새로운 토큰 발급 요청...") self._last_token_request_time = datetime.now() # HTTP 요청도 Lock 안에서 실행 — 중복 발급 방지 try: 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"] # 토큰 만료 시간 설정 (일반적으로 24시간) self.token_expires_at = datetime.now() + timedelta(hours=23) # 파일에 저장 (비동기) — Lock 안에서 호출하되 내부는 to_thread이므로 OK await self._save_token_to_cache() logger.info(f"✅ 새 토큰 발급 완료 (만료: {self.token_expires_at.strftime('%Y-%m-%d %H:%M')})") return self.access_token else: error_msg = response.text logger.error(f"❌ 토큰 발급 실패: {response.status_code} - {error_msg}") # 토큰 발급 실패 시에도 기존 토큰이 있다면 계속 사용 시도 if self.access_token: logger.warning(f"⚠️ 토큰 발급 실패했지만 기존 토큰 재사용 시도") # 만료 시간 연장 self.token_expires_at = datetime.now() + timedelta(hours=1) return self.access_token raise Exception(f"Failed to get access token: {error_msg}") except Exception as e: logger.error(f"❌ 토큰 요청 중 오류: {e}") # 기존 토큰이 있으면 사용 if self.access_token: logger.warning(f"⚠️ 토큰 요청 실패, 기존 토큰 재사용") self.token_expires_at = datetime.now() + timedelta(hours=1) return self.access_token raise async def _make_request(self, method: str, endpoint: str, tr_id: str, extra_headers: Optional[Dict[str, str]] = None, **kwargs) -> (Dict[str, Any], Any): """API 요청 실행 (Circuit Breaker + 세마포어 + 재시도 + 모니터링)""" # Circuit Breaker 상태 확인 if not self.circuit_breaker.can_request(): wait_sec = self.circuit_breaker.recovery_timeout logger.warning(f"⚡ Circuit breaker OPEN — {wait_sec}초 후 재시도 가능. 요청 거부: {endpoint}") raise Exception(f"Circuit breaker open: KIS API 일시 차단 ({wait_sec}s)") # 동시 요청 수 제한 (세마포어) async with self.semaphore: return await self._make_request_with_retry(method, endpoint, tr_id, extra_headers=extra_headers, **kwargs) async def _make_request_with_retry(self, method: str, endpoint: str, tr_id: str, extra_headers: Optional[Dict[str, str]] = None, **kwargs) -> (Dict[str, Any], Any): """재시도 로직을 포함한 실제 요청 실행""" start_time = time.time() retry_count = 0 last_exception = None 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: duration = time.time() - start_time self.monitor.record_call(success=True, duration=duration, retries=retry_count) self.circuit_breaker.record_success() # 오류율 경고 if self.monitor.should_alert(threshold=settings.API_ERROR_THRESHOLD): logger.warning( f"⚠️ API 오류율 높음: {self.monitor.get_error_rate():.1%} " f"(임계값: {settings.API_ERROR_THRESHOLD:.1%})" ) return response.json(), response.headers # API 호출 제한 초과 (429) 또는 서버 오류 (5xx) elif response.status_code in [429, 500, 502, 503, 504]: retry_count += 1 last_exception = Exception( f"API rate limit or server error: {response.status_code} - {response.text}" ) if attempt < self.max_retries: # 지수 백오프 적용 delay = self.retry_delay * (self.retry_backoff ** attempt) logger.warning( f"⏳ API 재시도 {attempt + 1}/{self.max_retries}: " f"상태코드 {response.status_code}, {delay:.1f}초 후 재시도" ) await asyncio.sleep(delay) continue else: self.circuit_breaker.record_failure() logger.error(f"❌ API 최대 재시도 횟수 초과: {response.status_code}") raise last_exception # 인증 오류 또는 잘못된 요청 (4xx) - 재시도 불필요 elif 400 <= response.status_code < 500: duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) error_msg = f"API 클라이언트 오류 (재시도 안함): {response.status_code} - {response.text}" logger.error(f"❌ {error_msg}") # 에러 응답도 json과 headers 튜플로 반환 시도 try: return response.json(), response.headers except Exception: return {"rt_cd": str(response.status_code), "msg1": error_msg, "output": None}, None # 기타 오류 else: duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) error_msg = f"API 요청 실패: {response.status_code} - {response.text}" logger.error(f"❌ {error_msg}") raise Exception(error_msg) except httpx.TimeoutException as e: retry_count += 1 last_exception = e if attempt < self.max_retries: delay = self.retry_delay * (self.retry_backoff ** attempt) logger.warning(f"⏳ API 타임아웃, {delay:.1f}초 후 재시도 ({attempt + 1}/{self.max_retries})") await asyncio.sleep(delay) continue else: duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) self.circuit_breaker.record_failure() logger.error(f"❌ API 타임아웃 최대 재시도 초과") raise except Exception as e: # 예상치 못한 오류 duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) self.circuit_breaker.record_failure() logger.error(f"❌ API 요청 중 예외 발생: {str(e)}") raise # 여기까지 왔다면 모든 재시도 실패 duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) self.circuit_breaker.record_failure() if last_exception: raise last_exception raise Exception("API 요청 실패 (알 수 없는 오류)") async def get_current_price(self, stock_code: str) -> Dict[str, Any]: """현재가 조회""" try: result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-price", "FHKST01010100", params={ "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code } ) return result except Exception as e: logger.error(f"Error fetching price for {stock_code}: {e}") raise async def get_daily_chart(self, stock_code: str, period: str = "D", limit: int = 240) -> Dict[str, Any]: """일봉/주봉/월봉 조회 (Redis 캐시 적용, 페이지네이션 지원으로 최대 240개까지 조회) 캐시 키: chart:daily:{stock_code}:{period}:{limit} TTL: 5분 (장중 업데이트 주기 고려) """ cache_key = f"chart:daily:{stock_code}:{period}:{limit}" # ── 1. 캐시 히트 확인 ─────────────────────────────────────────────── cached = await _cache_get(cache_key) if cached is not None: logger.debug(f"📦 [캐시 히트] {cache_key}") return cached # ── 2. KIS API 호출 ───────────────────────────────────────────────── all_output2 = [] params = { "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code, "FID_PERIOD_DIV_CODE": period, "FID_ORG_ADJ_PRC": "0" # 0: 수정주가, 1: 원주가 } try: result, headers = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-daily-price", "FHKST01010400", params=params ) data_key = 'output2' if isinstance(result.get('output2'), list) and result.get('output2') else \ 'output' if isinstance(result.get('output'), list) and result.get('output') else None if not result or result.get('rt_cd') != '0' or data_key is None: logger.warning(f"No chart data for {stock_code}: {result.get('msg1', 'Unknown')}") return result all_output2.extend(result.get(data_key, [])) tr_cont = headers.get('tr_cont') # 연속 조회 while tr_cont in ('F', 'M') and len(all_output2) < limit: await asyncio.sleep(self.min_request_interval + 0.1) extra_headers = {'tr_cont': 'Y'} result_cont, headers_cont = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-daily-price", "FHKST01010400", params=params, extra_headers=extra_headers ) if result_cont and result_cont.get('rt_cd') == '0': cont_key = 'output2' if isinstance(result_cont.get('output2'), list) else 'output' all_output2.extend(result_cont.get(cont_key, [])) tr_cont = headers_cont.get('tr_cont') if headers_cont else None final_result = result final_result['output2'] = all_output2 final_result['msg1'] = f"OK (fetched {len(all_output2)} data points)" logger.info(f"✅ Fetched {len(all_output2)} chart data points for {stock_code}") # ── 3. 캐시 저장 ──────────────────────────────────────────────── await _cache_set(cache_key, TTL_DAILY_CHART, final_result) logger.debug(f"💾 [캐시 저장] {cache_key} TTL={TTL_DAILY_CHART}s") return final_result except Exception as e: logger.error(f"Error fetching paginated chart for {stock_code}: {e}") raise async def get_daily_ohlcv_range( self, stock_code: str, start_date: str, end_date: str ) -> list: """날짜 범위 일봉 OHLCV 전체 조회 (FHKST03010100) KIS API는 1회 최대 100행 반환 → 날짜를 소급하며 start_date까지 전부 수집. Args: stock_code: 종목코드 (예: "005930") start_date: 조회 시작일 YYYYMMDD end_date: 조회 종료일 YYYYMMDD Returns: [{"stck_bsop_date": "20240101", "stck_oprc": ..., "stck_hgpr": ..., "stck_lwpr": ..., "stck_clpr": ..., "acml_vol": ..., "prdy_ctrt": ...}, ...] 최신 → 오래된 순 정렬 (KIS API 기본값) """ all_data: list = [] current_end = end_date while True: params = { "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code, "FID_INPUT_DATE_1": start_date, "FID_INPUT_DATE_2": current_end, "FID_PERIOD_DIV_CODE": "D", "FID_ORG_ADJ_PRC": "0", # 수정주가 } try: result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice", "FHKST03010100", params=params, ) except Exception as exc: logger.error( "get_daily_ohlcv_range %s %s~%s 실패: %s", stock_code, start_date, current_end, exc, ) break if not result or result.get("rt_cd") != "0": logger.warning( "get_daily_ohlcv_range %s: %s", stock_code, result.get("msg1") if result else "no response", ) break output = result.get("output2") or [] if not output: break all_data.extend(output) # 가장 오래된 행의 날짜를 확인하고 그보다 하루 전으로 소급 oldest_date = output[-1].get("stck_bsop_date", "") if not oldest_date or oldest_date <= start_date: break oldest_dt = datetime.strptime(oldest_date, "%Y%m%d") current_end = (oldest_dt - timedelta(days=1)).strftime("%Y%m%d") if current_end < start_date: break await asyncio.sleep(self.min_request_interval) return all_data async def get_overseas_price(self, symbol: str, exchange_code: str = "NASD") -> Dict[str, Any]: """해외주식 현재가 조회 (NASD, NYSE, AMEX)""" try: exch_map = {"NASD": "NAS", "NYSE": "NYS", "AMEX": "AMS"} formatted_exchange = exch_map.get(exchange_code.upper(), exchange_code.upper()) result, _ = await self._make_request( "GET", "/uapi/overseas-price/v1/quotations/price", "HHDFS00000300", params={ "AUTH": "", "EXCD": exchange_code.upper(), "SYMB": symbol.upper() } ) return result except Exception as e: logger.error(f"Error fetching overseas price for {symbol}: {e}") raise async def get_minute_chart( self, stock_code: str, ref_time: str = "", include_past: bool = True, ) -> Dict[str, Any]: """국내주식 분봉차트 조회 (FHKST03010200, Redis 캐시 적용) 캐시 키: chart:minute:{stock_code}:{ref_time_bucket} - ref_time_bucket: ref_time을 1분 단위로 버킷화 (HHMM) TTL: 1분 KIS API: /uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice 한 번 호출에 최대 30개 분봉 반환. Args: stock_code: 종목코드 (e.g. "005930") ref_time: 조회 기준 시간 (HHMMSS). 빈 문자열이면 현재 시간 include_past: True = 과거 데이터 포함 (Y), False = 당일만 (N) Returns: { "output1": { 현재가 요약 }, "output2": [ { "stck_bsop_date": "20260222", "stck_cntg_hour": "091500", "stck_prpr": "71800", "stck_oprc": "71500", "stck_hgpr": "72000", "stck_lwpr": "71400", "cntg_vol": "12345", "acml_vol": "987654", "acml_tr_pbmn": "70896123456", }, ... ] } """ try: if not ref_time: from datetime import datetime import pytz kst = pytz.timezone("Asia/Seoul") ref_time = datetime.now(kst).strftime("%H%M%S") # 캐시 키: 분 단위 버킷 (HHMM) — 같은 분 안에서 중복 호출 방지 ref_bucket = ref_time[:4] # HHMM cache_key = f"chart:minute:{stock_code}:{ref_bucket}" # ── 1. 캐시 히트 확인 ───────────────────────────────────────── cached = await _cache_get(cache_key) if cached is not None: logger.debug(f"📦 [캐시 히트] {cache_key}") return cached # ── 2. KIS API 호출 ─────────────────────────────────────────── params = { "FID_ETC_CLS_CODE": "", "FID_INPUT_ISCD": stock_code, "FID_INPUT_HOUR_1": ref_time, "FID_PW_DATA_INCU_YN": "Y" if include_past else "N", } result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice", "FHKST03010200", params=params, ) # ── 3. 캐시 저장 ────────────────────────────────────────────── await _cache_set(cache_key, TTL_MINUTE_CHART, result) logger.debug(f"💾 [캐시 저장] {cache_key} TTL={TTL_MINUTE_CHART}s") return result except Exception as e: logger.error(f"Error fetching minute chart for {stock_code}: {e}") raise async def get_minute_chart_full_day( self, stock_code: str, market_open: str = "090000", ) -> list: """당일 전체 분봉 데이터 수집 (Redis 캐시 적용, 여러 번 호출 집계) 캐시 키: chart:minute_fullday:{stock_code}:{YYYYMMDD_HHMM} - HHMM: 현재 시각의 분 단위 버킷 → 1분마다 갱신 TTL: 1분 09:00부터 현재까지의 모든 1분봉을 순서대로 반환. 한 번 호출에 30개씩 → 최대 ~13회 호출 필요 (390분/30). Returns: List[dict] — output2 레코드 목록 (오래된 것 → 최신 순) """ from datetime import datetime import pytz kst = pytz.timezone("Asia/Seoul") now_dt = datetime.now(kst) today_str = now_dt.strftime("%Y%m%d") now_bucket = now_dt.strftime("%H%M") # 1분 단위 버킷 cache_key = f"chart:minute_fullday:{stock_code}:{today_str}_{now_bucket}" # ── 1. 캐시 히트 확인 ─────────────────────────────────────────────── cached = await _cache_get(cache_key) if cached is not None: logger.debug(f"📦 [캐시 히트] {cache_key} ({len(cached)}개 분봉)") return cached # ── 2. KIS API 다중 호출 집계 ──────────────────────────────────────── now_str = now_dt.strftime("%H%M%S") all_candles: list = [] ref = now_str seen_times: set = set() for _ in range(15): # 최대 15회 (450분, 여유 있게) try: result = await self.get_minute_chart(stock_code, ref_time=ref) except Exception as e: logger.warning(f"Minute chart fetch failed for {stock_code} at {ref}: {e}") break candles = result.get("output2") if isinstance(result.get("output2"), list) else [] if not candles: break new_added = 0 for c in candles: t = c.get("stck_cntg_hour", "") if t and t not in seen_times: seen_times.add(t) all_candles.append(c) new_added += 1 times = [c.get("stck_cntg_hour", "999999") for c in candles if c.get("stck_cntg_hour")] oldest = min(times) if times else "000000" if oldest <= market_open or new_added == 0: break ref = oldest await asyncio.sleep(self.min_request_interval + 0.05) # 오래된 것 → 최신 순으로 정렬 all_candles.sort(key=lambda c: c.get("stck_cntg_hour", "")) # ── 3. 캐시 저장 ──────────────────────────────────────────────────── if all_candles: await _cache_set(cache_key, TTL_MINUTE_FULLDAY, all_candles) logger.debug(f"💾 [캐시 저장] {cache_key} ({len(all_candles)}개 분봉) TTL={TTL_MINUTE_FULLDAY}s") return all_candles async def get_overseas_daily_chart(self, symbol: str, exchange_code: str = "NASD", period: str = "D") -> Dict[str, Any]: """해외주식 차트 조회 (Redis 캐시 적용) 캐시 키: chart:overseas:{exchange_code}:{symbol}:{period} TTL: 5분 """ cache_key = f"chart:overseas:{exchange_code.upper()}:{symbol.upper()}:{period}" # ── 1. 캐시 히트 확인 ─────────────────────────────────────────────── cached = await _cache_get(cache_key) if cached is not None: logger.debug(f"📦 [캐시 히트] {cache_key}") return cached # ── 2. KIS API 호출 ───────────────────────────────────────────────── try: gubn = "0" if period == "W": gubn = "1" elif period == "M": gubn = "2" result, _ = await self._make_request( "GET", "/uapi/overseas-price/v1/quotations/dailyprice", "HHDFS76240000", params={ "AUTH": "", "EXCD": exchange_code.upper(), "SYMB": symbol.upper(), "GUBN": gubn, "BYMD": "", "MODP": "1" } ) # ── 3. 캐시 저장 ──────────────────────────────────────────────── await _cache_set(cache_key, TTL_OVERSEAS_CHART, result) logger.debug(f"💾 [캐시 저장] {cache_key} TTL={TTL_OVERSEAS_CHART}s") return result except Exception as e: logger.error(f"Error fetching overseas chart for {symbol}: {e}") raise async def place_overseas_order( self, symbol: str, order_type: str, # "BUY" or "SELL" quantity: int, price: float, # 해외주식은 지정가(Limit)가 기본 exchange_code: str = "NASD" ) -> Dict[str, Any]: """해외주식 주문 (미국주식 기준)""" try: account_number_prefix = self.account_number[:8] account_number_suffix = self.account_number[8:] # 모의/실전 구분 is_virtual = "vts" in self.base_url.lower() # TR_ID 설정 (미국 매수/매도) if order_type == "BUY": tr_id = "VTTT1002U" if is_virtual else "TTTT1002U" else: tr_id = "VTTT1006U" if is_virtual else "TTTT1006U" result, _ = await self._make_request( "POST", "/uapi/overseas-stock/v1/trading/order", tr_id, json={ "CANO": account_number_prefix, "ACNT_PRDT_CD": account_number_suffix, "OVRS_EXCG_CD": exchange_code.upper(), "PDNO": symbol.upper(), "ORD_QTY": str(quantity), "OVRS_ORD_UNPR": str(price), "ORD_SVR_DVSN_CD": "0", "ORD_DVSN": "00" } ) return result except Exception as e: logger.error(f"Error placing overseas order: {e}") raise async def get_balance(self) -> Dict[str, Any]: """계좌 잔고 조회""" try: account_number_prefix = self.account_number[:8] account_number_suffix = self.account_number[8:] # 모의투자 환경 확인 (URL에 'vts' 포함 여부로 판단) is_virtual = "vts" in self.base_url.lower() tr_id = "VTTC8434R" if is_virtual else "TTTC8434R" result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/trading/inquire-balance", tr_id, params={ "CANO": account_number_prefix, "ACNT_PRDT_CD": account_number_suffix, "AFHR_FLPR_YN": "N", "OFL_YN": "", "INQR_DVSN": "02", "UNPR_DVSN": "01", "FUND_STTL_ICLD_YN": "N", "FNCG_AMT_AUTO_RDPT_YN": "N", "PRCS_DVSN": "01", "CTX_AREA_FK100": "", "CTX_AREA_NK100": "" } ) logger.info(f"📋 Balance API Full Response: {result}") logger.info(f"📋 Balance API Keys: {result.keys()}") if result.get('rt_cd') != '0': logger.error(f"❌ Balance API error - rt_cd={result.get('rt_cd')}, msg1={result.get('msg1')}") return result except Exception as e: logger.error(f"Error fetching balance: {e}") raise async def get_overseas_balance(self) -> Dict[str, Any]: """해외주식 잔고 조회 (미국 기준) 실전: TTTS3012R / 모의: VTTS3012R 응답 output1: 보유 종목 목록 (ovrs_pdno, ovrs_item_name, ovrs_excg_cd, ord_psbl_qty, pchs_avg_pric, now_pric2, evlu_pfls_amt, evlu_pfls_rt2) 응답 output2: 계좌 요약 (tot_evlu_amt, frcr_evlu_tota, usd_amt, ovrs_tot_pfls) """ try: account_number_prefix = self.account_number[:8] account_number_suffix = self.account_number[8:] is_virtual = "vts" in self.base_url.lower() tr_id = "VTTS3012R" if is_virtual else "TTTS3012R" result, _ = await self._make_request( "GET", "/uapi/overseas-stock/v1/trading/inquire-balance", tr_id, params={ "CANO": account_number_prefix, "ACNT_PRDT_CD": account_number_suffix, "OVRS_EXCG_CD": "NASD", # 미국 전체 (NASD 조회 시 NYSE·AMEX 포함) "TR_CRCY_CD": "USD", "CTX_AREA_FK200": "", "CTX_AREA_NK200": "", } ) return result except Exception as e: logger.error(f"Error fetching overseas balance: {e}") return {} async def place_order( self, stock_code: str, order_type: str, # "BUY" or "SELL" quantity: int, price: Optional[float] = None # None for market order ) -> Dict[str, Any]: """주문 실행""" try: if not self.app_key or not self.app_secret or not self.account_number: error_msg = f"KIS API 설정 누락: app_key={bool(self.app_key)}, secret={bool(self.app_secret)}, account={bool(self.account_number)}" logger.error(error_msg) return { "rt_cd": "-1", "msg1": error_msg, "odno": None } account_number_prefix = self.account_number[:8] account_number_suffix = self.account_number[8:] # 모의투자 환경 확인 is_virtual = "vts" in self.base_url.lower() if order_type == "BUY": tr_id = "VTTC0802U" if is_virtual else "TTTC0802U" else: tr_id = "VTTC0801U" if is_virtual else "TTTC0801U" order_dvsn = "01" if price is None else "00" order_price = "0" if price is None else str(int(price)) logger.info(f"📤 주문 요청: {order_type} {stock_code} {quantity}주 @{order_price}원 (tr_id={tr_id}, is_virtual={is_virtual})") result, _ = await self._make_request( "POST", "/uapi/domestic-stock/v1/trading/order-cash", tr_id, json={ "CANO": account_number_prefix, "ACNT_PRDT_CD": account_number_suffix, "PDNO": stock_code, "ORD_DVSN": order_dvsn, "ORD_QTY": str(quantity), "ORD_UNPR": order_price } ) logger.info(f"📥 주문 응답: rt_cd={result.get('rt_cd')}, msg1={result.get('msg1')}, odno={result.get('odno')}") return result except Exception as e: logger.error(f"❌ 주문 실행 오류: {e}", exc_info=True) return { "rt_cd": "-1", "msg1": str(e), "odno": None } async def get_order_execution(self, order_id: str, is_domestic: bool = True) -> Dict[str, Any]: """주문 체결 내역 단건 조회. 국내: TTTC8001R (실전) / VTTC8001R (모의) Returns: {"filled_qty": int, "filled_price": float} 체결 미확인 시 filled_qty=0 반환. """ try: account_number_prefix = self.account_number[:8] account_number_suffix = self.account_number[8:] is_virtual = "vts" in self.base_url.lower() today = datetime.now().strftime("%Y%m%d") if is_domestic: tr_id = "VTTC8001R" if is_virtual else "TTTC8001R" result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/trading/inquire-ccnl", tr_id, params={ "CANO": account_number_prefix, "ACNT_PRDT_CD": account_number_suffix, "INQR_STRT_DT": today, "INQR_END_DT": today, "SLL_BUY_DVSN_CD": "00", "INQR_DVSN": "00", "PDNO": "", "CCLD_DVSN": "00", "ORD_GNO_BRNO": "", "ODNO": order_id, "INQR_DVSN_3": "00", "INQR_DVSN_1": "", "CTX_AREA_FK100": "", "CTX_AREA_NK100": "", }, ) output = result.get("output1", []) if output: row = output[0] if isinstance(output, list) else output filled_qty = int(row.get("tot_ccld_qty", 0) or 0) filled_price = float(row.get("avg_prvs", 0) or 0) return {"filled_qty": filled_qty, "filled_price": filled_price} # 해외 체결 조회는 별도 TR_ID 필요 — 현재 미지원 (포트폴리오 동기화로 대체) return {"filled_qty": 0, "filled_price": 0.0} except Exception as e: logger.debug(f"[get_order_execution] order_id={order_id} 조회 오류: {e}") return {"filled_qty": 0, "filled_price": 0.0} async def cancel_order(self, order_id: str, quantity: int) -> Dict[str, Any]: """주문 취소""" try: account_number_prefix = self.account_number[:8] account_number_suffix = self.account_number[8:] # 모의투자 환경 확인 is_virtual = "vts" in self.base_url.lower() tr_id = "VTTC0803U" if is_virtual else "TTTC0803U" result, _ = await self._make_request( "POST", "/uapi/domestic-stock/v1/trading/order-rvsecncl", tr_id, json={ "CANO": account_number_prefix, "ACNT_PRDT_CD": account_number_suffix, "KRX_FWDG_ORD_ORGNO": "", "ORGN_ODNO": order_id, "ORD_DVSN": "00", "RVSE_CNCL_DVSN_CD": "02", # 취소 "ORD_QTY": "0", "ORD_UNPR": "0", "QTY_ALL_ORD_YN": "Y" } ) return result except Exception as e: logger.error(f"Error canceling order: {e}") raise async def search_stocks_by_condition( self, market: str = "J", condition_name: str = "", max_results: int = 100 ) -> Dict[str, Any]: """조건 검색으로 종목 조회""" try: result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-search", "HHKST03900300", params={ "FID_COND_MRKT_DIV_CODE": market, "FID_COND_SCR_DIV_CODE": "20171", "FID_INPUT_ISCD": "0000", "FID_DIV_CLS_CODE": "0", "FID_BLNG_CLS_CODE": "0", "FID_TRGT_CLS_CODE": "111111111", "FID_TRGT_EXLS_CLS_CODE": "000000", "FID_INPUT_PRICE_1": "", "FID_INPUT_PRICE_2": "", "FID_VOL_CNT": "", "FID_INPUT_DATE_1": "" } ) return result except Exception as e: logger.error(f"Error searching stocks: {e}") raise async def get_stock_list(self, market: str = "J") -> Dict[str, Any]: """전체 종목 리스트 조회""" try: result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/volume-rank", "FHPST01710000", params={ "FID_COND_MRKT_DIV_CODE": market, "FID_COND_SCR_DIV_CODE": "20171", "FID_INPUT_ISCD": "0000", "FID_DIV_CLS_CODE": "0", "FID_BLNG_CLS_CODE": "0", "FID_TRGT_CLS_CODE": "111111111", "FID_TRGT_EXLS_CLS_CODE": "000000", "FID_INPUT_PRICE_1": "5000", # 5,000원 이상 종목만 조회 "FID_INPUT_PRICE_2": "", "FID_VOL_CNT": "", "FID_INPUT_DATE_1": "" } ) return result except Exception as e: logger.error(f"Error getting stock list: {e}") raise async def get_volume_rank(self, market: str = "J") -> Dict[str, Any]: """거래량 순위 조회 (get_stock_list와 동일하지만 명시적 이름)""" return await self.get_stock_list(market) async def get_orderbook(self, stock_code: str) -> Dict[str, Any]: """호가 정보 조회 (매수/매도 호가 10단계)""" try: result, _ = await self._make_request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn", "FHKST01010200", params={ "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code } ) logger.info(f"📊 Orderbook raw response for {stock_code}:") logger.info(f" rt_cd: {result.get('rt_cd')}, msg1: {result.get('msg1')}") if result.get('output1'): logger.info(f" output1 keys: {list(result['output1'].keys())[:10]}") if result.get('output2'): logger.info(f" output2 keys: {list(result['output2'].keys())[:10]}") return result except Exception as e: logger.error(f"Error fetching orderbook for {stock_code}: {e}") raise def get_monitor_stats(self) -> Dict[str, Any]: """API 호출 모니터링 통계 반환""" return self.monitor.get_stats() def log_monitor_stats(self): """모니터링 통계 로깅""" stats = self.get_monitor_stats() logger.info( f"📊 API 통계: 총 {stats['total_calls']}건, " f"오류 {stats['total_errors']}건 ({stats['error_rate']:.1%}), " f"재시도 {stats['total_retries']}건, " f"평균 응답 {stats['avg_duration']:.2f}초" ) # 싱글톤 인스턴스 korea_investment_api = KoreaInvestmentAPI() # 실계좌 시세조회 전용 싱글톤 (KIS_REAL_APP_KEY 설정 시 초기화) # 모의투자 모드에서도 시세/차트/종목 조회는 실계좌 API를 사용한다. try: real_market_api = KoreaInvestmentAPI.for_real_market_data() except Exception: real_market_api = None