diff --git a/backend/app/trading/kis_client.py b/backend/app/trading/kis_client.py new file mode 100644 index 0000000..f237f90 --- /dev/null +++ b/backend/app/trading/kis_client.py @@ -0,0 +1,233 @@ +""" +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 diff --git a/backend/app/trading/order_manager.py b/backend/app/trading/order_manager.py new file mode 100644 index 0000000..e1920e9 --- /dev/null +++ b/backend/app/trading/order_manager.py @@ -0,0 +1,194 @@ +""" +Order Manager +주문 실행·취소·조회 전담 모듈 +""" +from typing import Dict, Any, Optional +import logging + +from app.trading.kis_client import kis_client + +logger = logging.getLogger(__name__) + + +class OrderManager: + """주문 실행 관리자""" + + def __init__(self, client=None): + self.client = client or kis_client + + async def place_order( + self, + stock_code: str, + order_type: str, # "BUY" or "SELL" + quantity: int, + price: Optional[float] = None # None = 시장가 + ) -> Dict[str, Any]: + """주문 실행""" + try: + account_number_prefix = self.client.account_number[:8] + account_number_suffix = self.client.account_number[8:] + + is_virtual = "vts" in self.client.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}") + + result, _ = await self.client.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')}, odno={result.get('odno')}") + return result + except Exception as e: + logger.error(f"❌ 주문 실패: {e}") + return {"rt_cd": "-1", "msg1": str(e), "odno": None} + + async def cancel_order(self, order_id: str, quantity: int) -> Dict[str, Any]: + """주문 취소""" + try: + account_number_prefix = self.client.account_number[:8] + account_number_suffix = self.client.account_number[8:] + + is_virtual = "vts" in self.client.base_url.lower() + tr_id = "VTTC0803U" if is_virtual else "TTTC0803U" + + result, _ = await self.client.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"❌ 취소 실패: {e}") + raise + + async def get_balance(self) -> Dict[str, Any]: + """계좌 잔고 조회""" + try: + account_number_prefix = self.client.account_number[:8] + account_number_suffix = self.client.account_number[8:] + + is_virtual = "vts" in self.client.base_url.lower() + tr_id = "VTTC8434R" if is_virtual else "TTTC8434R" + + result, _ = await self.client.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": "" + } + ) + return result + except Exception as e: + logger.error(f"❌ 잔고 조회 실패: {e}") + raise + + async def place_overseas_order( + self, + symbol: str, + order_type: str, + quantity: int, + price: float, + exchange_code: str = "NASD" + ) -> Dict[str, Any]: + """해외주식 주문""" + try: + account_number_prefix = self.client.account_number[:8] + account_number_suffix = self.client.account_number[8:] + + is_virtual = "vts" in self.client.base_url.lower() + + if order_type == "BUY": + tr_id = "VTTT1002U" if is_virtual else "TTTT1002U" + else: + tr_id = "VTTT1006U" if is_virtual else "TTTT1006U" + + result, _ = await self.client.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"❌ 해외주식 주문 실패: {e}") + raise + + async def get_overseas_balance(self) -> Dict[str, Any]: + """해외주식 잔고 조회""" + try: + account_number_prefix = self.client.account_number[:8] + account_number_suffix = self.client.account_number[8:] + + is_virtual = "vts" in self.client.base_url.lower() + tr_id = "VTTS3012R" if is_virtual else "TTTS3012R" + + result, _ = await self.client.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", + "TR_CRCY_CD": "USD", + "CTX_AREA_FK200": "", + "CTX_AREA_NK200": "", + } + ) + return result + except Exception as e: + logger.error(f"❌ 해외주식 잔고 조회 실패: {e}") + return {} + + +# 싱글톤 +order_manager = OrderManager()