initial commit

한국투자증권(API)/examples_llm/auth/auth_token/auth_token.py

@@ -0,0 +1,120 @@
+# [인증] OAuth 접근토큰 발급
+# Generated by KIS API Generator (Single API Mode)
+# -*- coding: utf-8 -*-
+"""
+Created on 2025-06-19
+
+"""
+
+import json
+import logging
+import sys
+
+import pandas as pd
+import requests
+
+sys.path.extend(['../..', '.'])
+import kis_auth as ka
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+##############################################################################################
+# [인증] OAuth 접근토큰 발급
+##############################################################################################
+
+# 상수 정의
+API_URL = "/oauth2/tokenP"
+
+def auth_token(
+        grant_type: str,
+        appkey: str,
+        appsecret: str,
+        env_dv: str
+) -> pd.DataFrame:
+    """
+    OAuth 접근토큰 발급 API를 호출하여 DataFrame으로 반환합니다.
+    
+    Args:
+        grant_type (str): [필수] 권한부여 Type (client_credentials)
+        appkey (str): [필수] 앱키 (한국투자증권 홈페이지에서 발급받은 appkey)
+        appsecret (str): [필수] 앱시크릿키 (한국투자증권 홈페이지에서 발급받은 appsecret)
+        env_dv (str): [필수] 환경구분 (real: 실전, demo: 모의)
+        
+    Returns:
+        pd.DataFrame: OAuth 토큰 발급 결과
+        
+    Example:
+        >>> df = auth_token(
+        ...     grant_type="client_credentials",
+        ...     appkey=trenv.my_app,
+        ...     appsecret=trenv.my_sec,
+        ...     env_dv="real"
+        ... )
+        >>> print(df)
+    """
+    # 필수 파라미터 검증
+    if not grant_type:
+        logger.error("grant_type is required. (e.g. 'client_credentials')")
+        raise ValueError("grant_type is required. (e.g. 'client_credentials')")
+
+    if not appkey:
+        logger.error("appkey is required. (한국투자증권 홈페이지에서 발급받은 appkey)")
+        raise ValueError("appkey is required. (한국투자증권 홈페이지에서 발급받은 appkey)")
+
+    if not appsecret:
+        logger.error("appsecret is required. (한국투자증권 홈페이지에서 발급받은 appsecret)")
+        raise ValueError("appsecret is required. (한국투자증권 홈페이지에서 발급받은 appsecret)")
+
+    if not env_dv:
+        logger.error("env_dv is required. (real: 실전, demo: 모의)")
+        raise ValueError("env_dv is required. (real: 실전, demo: 모의)")
+
+    # 환경 구분에 따른 서버 URL 설정
+    config = ka.getEnv()
+    if env_dv == "real":
+        base_url = config.get("prod", "")
+    elif env_dv == "demo":
+        base_url = config.get("vps", "")
+    else:
+        logger.error("env_dv must be 'real' or 'demo'")
+        raise ValueError("env_dv must be 'real' or 'demo'")
+    
+    url = f"{base_url}{API_URL}"
+    
+    # 헤더 설정
+    headers = {
+        "Content-Type": "application/json",
+        "Accept": "text/plain",
+        "charset": "UTF-8"
+    }
+    
+    # 요청 데이터
+    data = {
+        "grant_type": grant_type,
+        "appkey": appkey,
+        "appsecret": appsecret,
+    }
+
+    try:
+        # POST 방식으로 직접 API 호출
+        response = requests.post(url, data=json.dumps(data), headers=headers)
+        
+        if response.status_code == 200:
+            # 응답 데이터를 DataFrame으로 반환 (1개 row)
+            response_data = response.json()
+            current_data = pd.DataFrame([response_data])
+            
+            logger.info("OAuth 토큰 발급 성공")
+            return current_data
+        else:
+            logger.error("API call failed: %s - %s", response.status_code, response.text)
+            return pd.DataFrame()
+            
+    except requests.RequestException as e:
+        logger.error("Request failed: %s", str(e))
+        return pd.DataFrame()
+    except json.JSONDecodeError as e:
+        logger.error("JSON decode failed: %s", str(e))
+        return pd.DataFrame()