브랜치 분리 방식: A / B / C

A 선택 시 커밋 메시지: 위 초안 OK / 수정 / 직접 작성
작업 시점: 지금 / 운영 데이터 1~2일 쌓고 / 주말
This commit is contained in:
2026-05-05 21:04:17 +09:00
parent c2b2b711e0
commit f61c471aac
58 changed files with 803502 additions and 1430 deletions

339
test_kiwoom_token.py Normal file
View File

@@ -0,0 +1,339 @@
#!/usr/bin/env python3
"""
키움 OpenAPI 토큰 발급 진단 스크립트
====================================
DB(env_config 최신 스냅샷)에 저장된 키움 키 4종 (MOCK·REAL × KEY·SECRET)을
자동으로 읽어 모의/실전 도메인 각각에 ``POST /oauth2/token`` 을 던지고,
응답·키 메타정보(길이·공백포함·앞뒤 일부)를 한 번에 보여 준다.
키움 토큰 엔드포인트 (kis_ws.KiwoomTokenManager._request_new_token 와 동일):
- 실전: https://api.kiwoom.com/oauth2/token
- 모의: https://mockapi.kiwoom.com/oauth2/token
- body: {"grant_type":"client_credentials","appkey":..., "secretkey":...}
사용법
------
DB 키로 자동 테스트::
python3 test_kiwoom_token.py
인터랙티브 입력 (키에 특수문자/공백 있을 때 가장 안전)::
python3 test_kiwoom_token.py --manual
python3 test_kiwoom_token.py --manual --mode real
CLI 인자로 직접 입력 (키 값은 반드시 ``""`` 로 감싸세요)::
python3 test_kiwoom_token.py --key "<APPKEY>" --secret "<SECRET>" --mode mock
python3 test_kiwoom_token.py --key="<APPKEY>" --secret="<SECRET>" --mode real
자주 보는 에러 코드
-------------------
- 8001: App Key / Secret 검증 실패 → 키 자체가 틀렸거나 OpenAPI 미신청 / 만료
- 1700: 요청 개수 초과 (au10001 rate limit) → 1분 정도 기다렸다가 재시도
- 9999: 서버 일시 오류 → 재시도
이 스크립트는 토큰 발급만 하므로 계좌번호는 사용하지 않는다 (키움 시세 API 정책).
"""
from __future__ import annotations
import argparse
import getpass
import json
import sys
from pathlib import Path
from typing import Optional, Tuple
import requests
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
# ──────────────────────────────────────────────────────────────────────
# 도메인 / 모드
# ──────────────────────────────────────────────────────────────────────
DOMAINS = {
"mock": "mockapi.kiwoom.com",
"real": "api.kiwoom.com",
}
# ──────────────────────────────────────────────────────────────────────
# 키 메타정보 진단
# ──────────────────────────────────────────────────────────────────────
def _safe_preview(s: str, head: int = 8, tail: int = 4) -> str:
"""키를 가운데를 가린 형태로 표시 (앞8자 …끝4자)."""
if not s:
return "(empty)"
if len(s) <= head + tail:
return repr(s)
return f"{s[:head]}{s[-tail:]}"
def _has_whitespace(s: str) -> bool:
return any(c in s for c in (" ", "\t", "\n", "\r", "\v", "\f"))
def diagnose_key(label: str, value: Optional[str]) -> bool:
"""키 값 메타 출력. 사용 가능하면 True."""
v = (value or "")
raw_len = len(v)
stripped = v.strip()
has_ws = _has_whitespace(v)
print(f"{label:28s} len={raw_len:3d} preview={_safe_preview(stripped)}"
f" whitespace={'YES⚠ ' if has_ws else 'no '}"
f" {'(empty)' if not stripped else ''}")
return bool(stripped)
# ──────────────────────────────────────────────────────────────────────
# 토큰 발급
# ──────────────────────────────────────────────────────────────────────
def request_token(
appkey: str,
secretkey: str,
mode: str,
timeout: int = 10,
) -> Tuple[bool, dict]:
"""
키움 토큰 발급 단일 호출.
Returns:
(success, response_data_dict)
success=True 이면 ``response_data_dict`` 안에 token 이 들어 있음.
"""
if mode not in DOMAINS:
raise ValueError(f"mode must be 'mock' or 'real', got {mode!r}")
url = f"https://{DOMAINS[mode]}/oauth2/token"
payload = {
"grant_type": "client_credentials",
"appkey": appkey,
"secretkey": secretkey,
}
try:
resp = requests.post(url, json=payload, timeout=timeout)
except requests.RequestException as e:
return False, {"_exception": str(e), "_url": url}
try:
data = resp.json()
except ValueError:
return False, {"_status": resp.status_code, "_text": resp.text[:500]}
data["_status"] = resp.status_code
token = (data.get("token") or data.get("access_token") or "").strip()
return bool(token), data
def explain_error(data: dict) -> str:
"""에러 응답 → 사람이 읽는 진단 메시지."""
if "_exception" in data:
return f"네트워크 예외: {data['_exception']}"
code = str(data.get("return_code", "")).strip()
msg = str(data.get("return_msg", "")).strip()
if "8001" in msg or code == "3":
return ("[8001] App Key / Secret 검증 실패\n"
" → 키 자체가 잘못되었거나 OpenAPI 신청이 안 된 상태.\n"
" → openapi.kiwoom.com 에서 모의/실전 OpenAPI 신청 여부 확인.\n"
" → 키를 사이트에서 다시 복사해 DB 재등록 (앞뒤 공백 주의).")
if "1700" in msg or code == "5":
return ("[1700] 요청 개수 초과 (au10001 rate limit)\n"
" → 1분 정도 기다렸다가 재시도. 키는 정상일 수 있음.")
if "9999" in msg:
return "[9999] 서버 일시 오류 → 재시도"
return f"기타 오류: code={code} msg={msg}"
def run_test(appkey: str, secretkey: str, mode: str, label: str) -> None:
"""단일 모드 테스트 + 결과 출력."""
print(f"\n── [{label}] mode={mode} domain={DOMAINS[mode]} ──")
if not appkey or not secretkey:
print(f" ⏭ 키 없음 (appkey={'' if not appkey else 'OK'}, "
f"secretkey={'' if not secretkey else 'OK'}) — 건너뜀")
return
ok, data = request_token(appkey.strip(), secretkey.strip(), mode)
if ok:
token = data.get("token") or data.get("access_token")
exp_dt = data.get("expires_dt", "?")
exp_in = data.get("expires_in", "?")
print(f" ✅ 발급 성공! token={_safe_preview(token, 8, 4)} "
f"expires_dt={exp_dt} expires_in={exp_in}")
else:
status = data.pop("_status", "?")
print(f" ❌ 발급 실패 HTTP={status}")
print(f" 응답: {json.dumps(data, ensure_ascii=False)}")
print(f" 진단: {explain_error(data)}")
# ──────────────────────────────────────────────────────────────────────
# DB 키 로드
# ──────────────────────────────────────────────────────────────────────
def load_keys_from_db() -> dict:
"""
DB(env_config 최신 스냅샷)에서 키움 4개 키 + 레거시 폴백 + KIS_MOCK 로드.
"""
try:
from database import TradeDB # type: ignore
except ImportError as e:
print(f"⚠️ database 모듈 import 실패: {e}")
return {}
db_path = SCRIPT_DIR / "quant_bot.db"
try:
db = TradeDB(db_path=str(db_path))
except TypeError:
db = TradeDB()
latest = db.get_latest_env()
snap = (latest or {}).get("snapshot") or {}
db.close()
return {
"KIWOOM_APP_KEY_MOCK": str(snap.get("KIWOOM_APP_KEY_MOCK", "") or ""),
"KIWOOM_APP_SECRET_MOCK": str(snap.get("KIWOOM_APP_SECRET_MOCK", "") or ""),
"KIWOOM_APP_KEY_REAL": str(snap.get("KIWOOM_APP_KEY_REAL", "") or ""),
"KIWOOM_APP_SECRET_REAL": str(snap.get("KIWOOM_APP_SECRET_REAL", "") or ""),
# 레거시 폴백
"KIWOOM_APP_KEY": str(snap.get("KIWOOM_APP_KEY", "") or ""),
"KIWOOM_APP_SECRET": str(snap.get("KIWOOM_APP_SECRET", "") or ""),
"KIS_MOCK": str(snap.get("KIS_MOCK", "true") or "true"),
}
# ──────────────────────────────────────────────────────────────────────
# main
# ──────────────────────────────────────────────────────────────────────
def _prompt_manual_keys() -> Tuple[str, str]:
"""
인터랙티브 입력 — 키에 ``--`` ``=`` ``$`` ``"`` 같은 특수문자가 있어도
셸 escaping 영향 없이 안전하게 받음. secretkey 는 getpass 로 화면 가림.
"""
print("\n📝 인터랙티브 입력 (Enter 로 확정, secretkey 는 화면에 안 찍힘)")
appkey = input(" appkey : ").strip()
secretkey = getpass.getpass(" secretkey: ").strip()
if not appkey or not secretkey:
print("❌ 빈 값은 허용되지 않습니다.")
sys.exit(2)
return appkey, secretkey
def main() -> int:
p = argparse.ArgumentParser(
description="키움 OpenAPI 토큰 발급 진단",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
p.add_argument("--key", help='직접 입력할 appkey (값은 "" 로 감싸길 권장)')
p.add_argument("--secret", help='직접 입력할 secretkey (값은 "" 로 감싸길 권장)')
p.add_argument("--manual", action="store_true",
help="인터랙티브 입력 모드 (키에 특수문자 있을 때 권장)")
p.add_argument("--mode", choices=("mock", "real", "both"), default="both",
help="테스트 도메인 (default: both)")
args = p.parse_args()
# ── 인터랙티브 모드 ────────────────────────────────────────────
if args.manual:
print("=" * 70)
print("🔧 인터랙티브 입력 모드 (DB 무시)")
print("=" * 70)
appkey, secretkey = _prompt_manual_keys()
diagnose_key("입력 APPKEY", appkey)
diagnose_key("입력 SECRETKEY", secretkey)
modes = ("mock", "real") if args.mode == "both" else (args.mode,)
for m in modes:
run_test(appkey, secretkey, m, label=f"입력키→{m.upper()}")
return 0
# ── CLI 인자 직접 입력 모드 ─────────────────────────────────────
if args.key or args.secret:
if not (args.key and args.secret):
print("❌ --key 와 --secret 은 함께 지정해야 합니다.\n"
" 특수문자 때문에 까다롭다면 --manual 을 사용하세요.\n"
" 예시: python3 test_kiwoom_token.py --manual --mode real")
return 2
print("=" * 70)
print("🔧 CLI 인자 입력 모드 (DB 무시)")
print("=" * 70)
diagnose_key("입력 APPKEY", args.key)
diagnose_key("입력 SECRETKEY", args.secret)
modes = ("mock", "real") if args.mode == "both" else (args.mode,)
for m in modes:
run_test(args.key, args.secret, m, label=f"입력키→{m.upper()}")
return 0
# ── DB 자동 모드 ──────────────────────────────────────────────
print("=" * 70)
print("🔍 키움 토큰 진단 — DB env_config 최신 스냅샷")
print("=" * 70)
keys = load_keys_from_db()
if not keys:
print("❌ DB 에서 키를 불러오지 못했습니다.")
return 1
print(f"\nKIS_MOCK = {keys['KIS_MOCK']!r} "
f"(코드 정책: 시세는 항상 실키, 단 본 스크립트는 둘 다 시도)")
print("\n📦 DB에 저장된 키 메타:")
diagnose_key("KIWOOM_APP_KEY_MOCK", keys["KIWOOM_APP_KEY_MOCK"])
diagnose_key("KIWOOM_APP_SECRET_MOCK", keys["KIWOOM_APP_SECRET_MOCK"])
diagnose_key("KIWOOM_APP_KEY_REAL", keys["KIWOOM_APP_KEY_REAL"])
diagnose_key("KIWOOM_APP_SECRET_REAL", keys["KIWOOM_APP_SECRET_REAL"])
if keys["KIWOOM_APP_KEY"] or keys["KIWOOM_APP_SECRET"]:
print(" (레거시 폴백)")
diagnose_key("KIWOOM_APP_KEY", keys["KIWOOM_APP_KEY"])
diagnose_key("KIWOOM_APP_SECRET", keys["KIWOOM_APP_SECRET"])
# ── 호출 ─────────────────────────────────────────────────────
if args.mode in ("mock", "both"):
run_test(
keys["KIWOOM_APP_KEY_MOCK"],
keys["KIWOOM_APP_SECRET_MOCK"],
"mock",
label="MOCK키→모의도메인",
)
if args.mode in ("real", "both"):
run_test(
keys["KIWOOM_APP_KEY_REAL"],
keys["KIWOOM_APP_SECRET_REAL"],
"real",
label="REAL키→실전도메인",
)
# 레거시 키도 있으면 둘 다 찔러본다 (어느 도메인에서 통하는지 확인)
if keys["KIWOOM_APP_KEY"] and keys["KIWOOM_APP_SECRET"]:
if args.mode in ("mock", "both"):
run_test(
keys["KIWOOM_APP_KEY"], keys["KIWOOM_APP_SECRET"],
"mock", label="LEGACY키→모의도메인",
)
if args.mode in ("real", "both"):
run_test(
keys["KIWOOM_APP_KEY"], keys["KIWOOM_APP_SECRET"],
"real", label="LEGACY키→실전도메인",
)
print("\n" + "=" * 70)
print("✅ 진단 끝")
print("=" * 70)
print("팁:")
print(" • whitespace=YES 가 보이면 → DB 에 공백 섞임. 키 다시 등록 필요.")
print(" • 8001 만 계속 → openapi.kiwoom.com 에서 OpenAPI 신청 여부 확인.")
print(" • 1700 → 1분 후 재시도 (rate limit).")
print(" • 둘 다 실패해도 매매에는 영향 없음 — 갭보정만 비활성됨.")
return 0
if __name__ == "__main__":
sys.exit(main())