209 lines
8.0 KiB
Python
209 lines
8.0 KiB
Python
"""
|
|
해당 접근토큰발급관리 파일을 그대로 사용하시면 오류 발생할 수 있습니다.
|
|
토큰저장 및 발급 함수 등을 참고하시되 본인 환경에 맞게 코드 수정하셔서 사용하시기 바랍니다.
|
|
|
|
Created on Wed Feb 15 16:57:19 2023
|
|
|
|
@author: Administrator
|
|
"""
|
|
|
|
import time, copy
|
|
import yaml
|
|
import requests
|
|
import json
|
|
|
|
import os
|
|
|
|
import pandas as pd
|
|
|
|
from collections import namedtuple
|
|
from datetime import datetime
|
|
|
|
config_root = os.getcwd() + '\\'
|
|
# config_root = 'd:\\KIS\\config\\' # 토큰 파일이 저장될 폴더, 제3자가 찾지 어렵도록 경로 설정하시기 바랍니다.
|
|
#token_tmp = config_root + 'KIS000000' # 토큰 로컬저장시 파일 이름 지정, 파일이름을 토큰값이 유추가능한 파일명은 삼가바랍니다.
|
|
#token_tmp = config_root + 'KIS' + datetime.today().strftime("%Y%m%d%H%M%S") # 토큰 로컬저장시 파일명 년월일시분초
|
|
token_tmp = config_root + 'KIS' + datetime.today().strftime("%Y%m%d") # 토큰 로컬저장시 파일명 년월일
|
|
|
|
# 접근토큰 관리하는 파일 존재여부 체크, 없으면 생성
|
|
if os.path.exists(token_tmp) == False:
|
|
f = open(token_tmp, "w+")
|
|
|
|
# 앱키, 앱시크리트, 토큰, 계좌번호 등 저장관리, 자신만의 경로와 파일명으로 설정하시기 바랍니다.
|
|
# pip install PyYAML (패키지설치)
|
|
with open(config_root + 'kis_devlp.yaml', encoding='UTF-8') as f:
|
|
_cfg = yaml.load(f, Loader=yaml.FullLoader)
|
|
|
|
_TRENV = tuple()
|
|
_last_auth_time = datetime.now()
|
|
_autoReAuth = False
|
|
_DEBUG = False
|
|
_isPaper = False
|
|
|
|
# 기본 헤더값 정의
|
|
_base_headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "text/plain",
|
|
"charset": "UTF-8",
|
|
'User-Agent': _cfg['my_agent']
|
|
}
|
|
|
|
# 토큰 발급 받아 저장 (토큰값, 토큰 유효시간,1일, 6시간 이내 발급신청시는 기존 토큰값과 동일, 발급시 알림톡 발송)
|
|
def save_token(my_token, my_expired):
|
|
valid_date = datetime.strptime(my_expired, '%Y-%m-%d %H:%M:%S')
|
|
print('Save token date: ', valid_date)
|
|
with open(token_tmp, 'w', encoding='utf-8') as f:
|
|
f.write(f'token: {my_token}\n')
|
|
f.write(f'valid-date: {valid_date}\n')
|
|
|
|
|
|
# 토큰 확인 (토큰값, 토큰 유효시간_1일, 6시간 이내 발급신청시는 기존 토큰값과 동일, 발급시 알림톡 발송)
|
|
def read_token():
|
|
try:
|
|
# 토큰이 저장된 파일 읽기
|
|
with open(token_tmp, encoding='UTF-8') as f:
|
|
tkg_tmp = yaml.load(f, Loader=yaml.FullLoader)
|
|
|
|
# 토큰 만료 일,시간
|
|
exp_dt = datetime.strftime(tkg_tmp['valid-date'], '%Y-%m-%d %H:%M:%S')
|
|
# 현재일자,시간
|
|
now_dt = datetime.today().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
print('expire dt: ', exp_dt, ' vs now dt:', now_dt)
|
|
# 저장된 토큰 만료일자 체크 (만료일시 > 현재일시 인경우 보관 토큰 리턴)
|
|
if exp_dt > now_dt:
|
|
return tkg_tmp['token']
|
|
else:
|
|
# print('Need new token: ', tkg_tmp['valid-date'])
|
|
return None
|
|
except Exception as e:
|
|
# print('read token error: ', e)
|
|
return None
|
|
|
|
# 토큰 유효시간 체크해서 만료된 토큰이면 재발급처리
|
|
def _getBaseHeader():
|
|
if _autoReAuth: reAuth()
|
|
return copy.deepcopy(_base_headers)
|
|
|
|
|
|
# 가져오기 : 앱키, 앱시크리트, 종합계좌번호(계좌번호 중 숫자8자리), 계좌상품코드(계좌번호 중 숫자2자리), 토큰, 도메인
|
|
def _setTRENV(cfg):
|
|
nt1 = namedtuple('KISEnv', ['my_app', 'my_sec', 'my_acct', 'my_prod', 'my_token', 'my_url'])
|
|
d = {
|
|
'my_app': cfg['my_app'], # 앱키
|
|
'my_sec': cfg['my_sec'], # 앱시크리트
|
|
'my_acct': cfg['my_acct'], # 종합계좌번호(8자리)
|
|
'my_prod': cfg['my_prod'], # 계좌상품코드(2자리)
|
|
'my_token': cfg['my_token'], # 토큰
|
|
'my_url': cfg['my_url'] # 실전 도메인 (https://openapi.koreainvestment.com:9443)
|
|
} # 모의 도메인 (https://openapivts.koreainvestment.com:29443)
|
|
|
|
global _TRENV
|
|
_TRENV = nt1(**d)
|
|
|
|
|
|
def isPaperTrading(): # 모의투자 매매
|
|
return _isPaper
|
|
|
|
|
|
# 실전투자면 'prod', 모의투자면 'vps'를 셋팅 하시기 바랍니다.
|
|
def changeTREnv(token_key, svr='prod', product='01'):
|
|
cfg = dict()
|
|
|
|
global _isPaper
|
|
if svr == 'prod': # 실전투자
|
|
ak1 = 'my_app' # 실전투자용 앱키
|
|
ak2 = 'my_sec' # 실전투자용 앱시크리트
|
|
_isPaper = False
|
|
elif svr == 'vps': # 모의투자
|
|
ak1 = 'paper_app' # 모의투자용 앱키
|
|
ak2 = 'paper_sec' # 모의투자용 앱시크리트
|
|
_isPaper = True
|
|
|
|
cfg['my_app'] = _cfg[ak1]
|
|
cfg['my_sec'] = _cfg[ak2]
|
|
|
|
if svr == 'prod' and product == '01': # 실전투자 주식투자, 위탁계좌, 투자계좌
|
|
cfg['my_acct'] = _cfg['my_acct_stock']
|
|
elif svr == 'prod' and product == '03': # 실전투자 선물옵션(파생)
|
|
cfg['my_acct'] = _cfg['my_acct_future']
|
|
elif svr == 'vps' and product == '01': # 모의투자 주식투자, 위탁계좌, 투자계좌
|
|
cfg['my_acct'] = _cfg['my_paper_stock']
|
|
elif svr == 'vps' and product == '03': # 모의투자 선물옵션(파생)
|
|
cfg['my_acct'] = _cfg['my_paper_future']
|
|
|
|
cfg['my_prod'] = product
|
|
cfg['my_token'] = token_key
|
|
cfg['my_url'] = _cfg[svr]
|
|
|
|
_setTRENV(cfg)
|
|
|
|
|
|
def _getResultObject(json_data):
|
|
_tc_ = namedtuple('res', json_data.keys())
|
|
|
|
return _tc_(**json_data)
|
|
|
|
|
|
# Token 발급, 유효기간 1일, 6시간 이내 발급시 기존 token값 유지, 발급시 알림톡 무조건 발송
|
|
def auth(svr='prod', product='01', url=None):
|
|
p = {
|
|
"grant_type": "client_credentials",
|
|
}
|
|
# 개인 환경파일 "kis_devlp.yaml" 파일을 참조하여 앱키, 앱시크리트 정보 가져오기
|
|
# 개인 환경파일명과 위치는 고객님만 아는 위치로 설정 바랍니다.
|
|
if svr == 'prod': # 실전투자
|
|
ak1 = 'my_app' # 앱키 (실전투자용)
|
|
ak2 = 'my_sec' # 앱시크리트 (실전투자용)
|
|
elif svr == 'vps': # 모의투자
|
|
ak1 = 'paper_app' # 앱키 (모의투자용)
|
|
ak2 = 'paper_sec' # 앱시크리트 (모의투자용)
|
|
|
|
# 앱키, 앱시크리트 가져오기
|
|
p["appkey"] = _cfg[ak1]
|
|
p["appsecret"] = _cfg[ak2]
|
|
|
|
# 기존 발급된 토큰이 있는지 확인
|
|
saved_token = read_token() # 기존 발급 토큰 확인
|
|
print("saved_token: ", saved_token)
|
|
if saved_token is None: # 기존 발급 토큰 확인이 안되면 발급처리
|
|
url = f'{_cfg[svr]}/oauth2/tokenP'
|
|
res = requests.post(url, data=json.dumps(p), headers=_getBaseHeader()) # 토큰 발급
|
|
rescode = res.status_code
|
|
if rescode == 200: # 토큰 정상 발급
|
|
my_token = _getResultObject(res.json()).access_token # 토큰값 가져오기
|
|
my_expired= _getResultObject(res.json()).access_token_token_expired # 토큰값 만료일시 가져오기
|
|
save_token(my_token, my_expired) # 새로 발급 받은 토큰 저장
|
|
else:
|
|
print('Get Authentification token fail!\nYou have to restart your app!!!')
|
|
return
|
|
else:
|
|
my_token = saved_token # 기존 발급 토큰 확인되어 기존 토큰 사용
|
|
|
|
# 발급토큰 정보 포함해서 헤더값 저장 관리, API 호출시 필요
|
|
changeTREnv(f"Bearer {my_token}", svr, product)
|
|
|
|
_base_headers["authorization"] = _TRENV.my_token
|
|
_base_headers["appkey"] = _TRENV.my_app
|
|
_base_headers["appsecret"] = _TRENV.my_sec
|
|
|
|
global _last_auth_time
|
|
_last_auth_time = datetime.now()
|
|
|
|
if (_DEBUG):
|
|
print(f'[{_last_auth_time}] => get AUTH Key completed!')
|
|
|
|
|
|
# end of initialize, 토큰 재발급, 토큰 발급시 유효시간 1일
|
|
# 프로그램 실행시 _last_auth_time에 저장하여 유효시간 체크, 유효시간 만료시 토큰 발급 처리
|
|
def reAuth(svr='prod', product='01'):
|
|
n2 = datetime.now()
|
|
if (n2 - _last_auth_time).seconds >= 86400: # 유효시간 1일
|
|
auth(svr, product)
|
|
|
|
# 접근토큰발급 저장
|
|
auth()
|
|
# 접근토큰 조회
|
|
gettoken = read_token()
|
|
print(gettoken)
|