121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
# [인증] 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()
|