300 lines
13 KiB
Python
300 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_psearch.py — KIS 종목조건검색 API 응답 확인용 단독 스크립트
|
|
==================================================================
|
|
[국내주식] 시세분석
|
|
① psearch-title (HHKST03900300) — 서버 저장 조건식 목록 조회
|
|
② psearch-result (HHKST03900400) — 특정 조건식 현재 결과 조회
|
|
|
|
⚠️ KIS 종목조건검색은 **모의투자 미지원**이라 실키(KIS_APP_KEY_REAL /
|
|
KIS_APP_SECRET_REAL) + 실전 도메인으로만 호출 가능.
|
|
본 스크립트는 KIS_MOCK 값과 무관하게 항상 실전 키로만 호출한다.
|
|
|
|
이 스크립트의 목적:
|
|
- 두 API 의 **요청 (URL/헤더/params)** 과 **응답 (status / 응답헤더 /
|
|
원본 JSON)** 을 그대로 찍어서 필드명·구조를 사람이 직접 검증.
|
|
- condition_manager 가 파싱하기 전 원본 응답을 보고 매핑 정확도 점검.
|
|
|
|
사용:
|
|
# 1) 목록만 보고 끝내기
|
|
$ python test_psearch.py --list
|
|
|
|
# 2) 목록 출력 후 seq 인터랙티브 입력
|
|
$ python test_psearch.py
|
|
|
|
# 3) 특정 seq 결과까지 한 번에
|
|
$ python test_psearch.py --seq 0
|
|
$ python test_psearch.py --seq 3 --user-id mylogin # HTS ID 임시 지정
|
|
|
|
옵션:
|
|
--seq <번호> 결과조회 대상 조건식 번호 (없으면 인터랙티브)
|
|
--user-id <ID> KIS_HTS_ID 대신 임시 지정 (테스트용)
|
|
--list 목록(psearch-title)만 호출하고 종료
|
|
--raw 응답 JSON 을 indent 없이 한 줄로 출력 (grep 용이)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
# kis_trader 패키지의 검증된 헬퍼들을 그대로 재사용 (별도 토큰/SafeRequest 구현 X)
|
|
from kis_trader.execution.kis_client import KISClient # noqa: E402
|
|
from kis_trader.utils.env import get_env_from_db # noqa: E402
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────
|
|
# 출력 헬퍼
|
|
# ──────────────────────────────────────────────────────────────────
|
|
def _mask(s: str, head: int = 6, tail: int = 4) -> str:
|
|
"""앱키 같은 시크릿을 로그에 찍을 때 앞뒤만 남기고 가린다."""
|
|
if not s:
|
|
return "(empty)"
|
|
if len(s) <= head + tail:
|
|
return "*" * len(s)
|
|
return f"{s[:head]}…{s[-tail:]}"
|
|
|
|
|
|
def _print_section(title: str) -> None:
|
|
print()
|
|
print("=" * 78)
|
|
print(f" {title}")
|
|
print("=" * 78)
|
|
|
|
|
|
def _dump_request(url: str, headers: dict, params: dict) -> None:
|
|
"""요청 정보 출력. authorization/appkey/appsecret 은 마스킹."""
|
|
print(f"[REQUEST]")
|
|
print(f" URL : {url}")
|
|
print(f" PARAMS : {json.dumps(params, ensure_ascii=False)}")
|
|
safe_h = dict(headers)
|
|
if "authorization" in safe_h:
|
|
tok = safe_h["authorization"].replace("Bearer ", "")
|
|
safe_h["authorization"] = "Bearer " + _mask(tok, 8, 6)
|
|
if "appkey" in safe_h:
|
|
safe_h["appkey"] = _mask(safe_h["appkey"])
|
|
if "appsecret" in safe_h:
|
|
safe_h["appsecret"] = _mask(safe_h["appsecret"])
|
|
print(f" HEADERS: {json.dumps(safe_h, ensure_ascii=False)}")
|
|
|
|
|
|
def _dump_response(r, raw: bool) -> dict:
|
|
"""응답 정보 출력 + 파싱한 dict 반환 (실패 시 빈 dict)."""
|
|
print(f"[RESPONSE]")
|
|
print(f" STATUS : {r.status_code}")
|
|
interesting_resp_headers = {
|
|
k: v for k, v in r.headers.items()
|
|
if k.lower() in (
|
|
"content-type", "tr_id", "tr_cont", "gt_uid",
|
|
"rate-limit-remaining", "rate-limit-reset",
|
|
)
|
|
}
|
|
if interesting_resp_headers:
|
|
print(f" HEADERS: {json.dumps(interesting_resp_headers, ensure_ascii=False)}")
|
|
|
|
try:
|
|
j = r.json()
|
|
except Exception as e:
|
|
print(f" BODY : <JSON 파싱 실패: {e}>")
|
|
print(f" TEXT : {r.text[:1000]}")
|
|
return {}
|
|
|
|
print(f" rt_cd : {j.get('rt_cd')} (0=정상, 그 외=에러)")
|
|
print(f" msg_cd : {j.get('msg_cd')}")
|
|
print(f" msg1 : {j.get('msg1')}")
|
|
print()
|
|
print("[RAW BODY]")
|
|
if raw:
|
|
print(json.dumps(j, ensure_ascii=False))
|
|
else:
|
|
print(json.dumps(j, ensure_ascii=False, indent=2))
|
|
return j
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────
|
|
# 핵심: 두 API 호출 (raw)
|
|
# ──────────────────────────────────────────────────────────────────
|
|
def call_psearch_title(client: KISClient, user_id: str, raw: bool) -> list:
|
|
"""
|
|
psearch-title (HHKST03900300) 호출 + 응답 출력.
|
|
Returns: parsed output2 list (사용자가 보기 쉽게 seq/이름 표 만들기용)
|
|
"""
|
|
_print_section("① psearch-title — 서버 저장 조건식 목록")
|
|
path = "/uapi/domestic-stock/v1/quotations/psearch-title"
|
|
tr_id = "HHKST03900300"
|
|
params = {"user_id": user_id}
|
|
headers = client._headers(tr_id)
|
|
url = client.base_url + path
|
|
|
|
_dump_request(url, headers, params)
|
|
r = client._get(path, tr_id, params)
|
|
j = _dump_response(r, raw)
|
|
|
|
out2 = j.get("output2") or []
|
|
if isinstance(out2, dict):
|
|
out2 = [out2]
|
|
|
|
if out2:
|
|
print()
|
|
print("[요약 — 사용자에게 의미 있는 필드만]")
|
|
print(f" 총 {len(out2)}건")
|
|
# KIS 실제 응답 키: condition_nm / grp_nm
|
|
# (문서엔 condition_name 처럼 적힌 자료가 많아 폴백 후보 다중 적용)
|
|
print(" ┌─────┬──────────────────────────────┬────────────────────┐")
|
|
print(" │ seq │ condition_nm │ grp_nm │")
|
|
print(" ├─────┼──────────────────────────────┼────────────────────┤")
|
|
for row in out2:
|
|
seq = str(row.get("seq", "?"))[:3].rjust(3)
|
|
nm = (
|
|
row.get("condition_nm")
|
|
or row.get("condition_name")
|
|
or row.get("cond_nm")
|
|
or ""
|
|
)
|
|
grp = (row.get("grp_nm") or row.get("group_nm") or "")
|
|
print(
|
|
f" │ {seq} │ {str(nm)[:28].ljust(28)} │ "
|
|
f"{str(grp)[:18].ljust(18)} │"
|
|
)
|
|
print(" └─────┴──────────────────────────────┴────────────────────┘")
|
|
return out2
|
|
|
|
|
|
def call_psearch_result(
|
|
client: KISClient, user_id: str, seq: str, raw: bool,
|
|
) -> list:
|
|
"""
|
|
psearch-result (HHKST03900400) 호출 + 응답 출력.
|
|
Returns: parsed output2 list
|
|
"""
|
|
_print_section(f"② psearch-result — 조건식 결과 (seq={seq})")
|
|
path = "/uapi/domestic-stock/v1/quotations/psearch-result"
|
|
tr_id = "HHKST03900400"
|
|
params = {"user_id": user_id, "seq": str(seq)}
|
|
headers = client._headers(tr_id)
|
|
url = client.base_url + path
|
|
|
|
_dump_request(url, headers, params)
|
|
r = client._get(path, tr_id, params)
|
|
j = _dump_response(r, raw)
|
|
|
|
out2 = j.get("output2") or []
|
|
if isinstance(out2, dict):
|
|
out2 = [out2]
|
|
|
|
if out2:
|
|
print()
|
|
print("[요약 — 가능한 키 후보 모두 시도해서 추출]")
|
|
print(f" 총 {len(out2)}종목")
|
|
print(" ┌────────┬──────────────────────────────┬──────────────┐")
|
|
print(" │ code │ name │ price │")
|
|
print(" ├────────┼──────────────────────────────┼──────────────┤")
|
|
for it in out2[:50]: # 상위 50개만
|
|
code = (
|
|
it.get("code") or it.get("stck_shrn_iscd")
|
|
or it.get("mksc_shrn_iscd") or ""
|
|
)
|
|
name = (
|
|
it.get("name") or it.get("hts_kor_isnm") or ""
|
|
)
|
|
price = (
|
|
it.get("stck_prpr") or it.get("price") or ""
|
|
)
|
|
print(
|
|
f" │ {str(code)[:6].ljust(6)} │ "
|
|
f"{str(name)[:28].ljust(28)} │ "
|
|
f"{str(price)[:12].rjust(12)} │"
|
|
)
|
|
if len(out2) > 50:
|
|
print(f" …(이하 {len(out2) - 50}종목 생략)")
|
|
print(" └────────┴──────────────────────────────┴──────────────┘")
|
|
return out2
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────
|
|
# 메인
|
|
# ──────────────────────────────────────────────────────────────────
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="KIS 종목조건검색(psearch-title / psearch-result) 응답 확인",
|
|
)
|
|
parser.add_argument("--seq", help="조건식 번호 (지정 시 결과조회까지 자동 실행)")
|
|
parser.add_argument("--user-id", help="KIS_HTS_ID 대신 임시 지정")
|
|
parser.add_argument("--list", action="store_true", help="목록만 호출하고 종료")
|
|
parser.add_argument("--raw", action="store_true", help="응답 JSON 을 한 줄로 출력")
|
|
args = parser.parse_args()
|
|
|
|
# ── HTS ID 결정 ──
|
|
user_id = (args.user_id or get_env_from_db("KIS_HTS_ID", "") or "").strip()
|
|
if not user_id:
|
|
print("❌ KIS_HTS_ID 미설정.")
|
|
print(" update_env_simple.py 로 KIS_HTS_ID 를 DB 에 저장하거나,")
|
|
print(" --user-id <ID> 옵션으로 임시 지정하세요.")
|
|
return 1
|
|
|
|
# ── 실키 검증 (모의 미지원) ──
|
|
real_key = (get_env_from_db("KIS_APP_KEY_REAL", "") or "").strip()
|
|
real_secret = (get_env_from_db("KIS_APP_SECRET_REAL", "") or "").strip()
|
|
if not real_key or not real_secret:
|
|
print("❌ KIS_APP_KEY_REAL / KIS_APP_SECRET_REAL 미설정.")
|
|
print(" psearch 는 모의투자 미지원이라 실키가 필수입니다.")
|
|
return 1
|
|
|
|
print(f"🔑 KIS_APP_KEY_REAL : {_mask(real_key)}")
|
|
print(f"🔑 KIS_APP_SECRET : {_mask(real_secret)}")
|
|
print(f"🪪 KIS_HTS_ID : {user_id}")
|
|
|
|
# ── 실전 도메인 KISClient (mock=False 강제) ──
|
|
try:
|
|
client = KISClient(
|
|
mock=False,
|
|
app_key=real_key,
|
|
app_secret=real_secret,
|
|
account_no=(get_env_from_db("KIS_ACCOUNT_NO_REAL", "") or "").strip(),
|
|
account_code=(get_env_from_db("KIS_ACCOUNT_CODE_REAL", "01") or "01").strip(),
|
|
)
|
|
except Exception as e:
|
|
print(f"❌ KISClient 초기화 실패: {e}")
|
|
return 1
|
|
|
|
if not client._token:
|
|
print("❌ 접근토큰 발급 실패. kis_token_manager 로그/실키 유효성을 확인하세요.")
|
|
return 1
|
|
print(f"🎫 access_token : {_mask(client._token, 10, 6)}")
|
|
print(f"🌐 base_url : {client.base_url}")
|
|
|
|
# ── ① 조건식 목록 ──
|
|
titles = call_psearch_title(client, user_id, args.raw)
|
|
|
|
if args.list:
|
|
return 0
|
|
|
|
# ── seq 결정 ──
|
|
seq = (args.seq or "").strip() if args.seq else ""
|
|
if not seq:
|
|
if not titles:
|
|
print()
|
|
print("⚠️ 조건식 목록이 비어 결과조회 단계로 진행할 수 없습니다.")
|
|
print(" HTS/MTS 에 먼저 조건식을 저장한 뒤 다시 실행하세요.")
|
|
return 0
|
|
print()
|
|
try:
|
|
seq = input(f"조회할 seq 입력 (예: {titles[0].get('seq', '0')}, 비우면 종료) > ").strip()
|
|
except EOFError:
|
|
seq = ""
|
|
if not seq:
|
|
print("⏹ 사용자 취소")
|
|
return 0
|
|
|
|
# ── ② 조건식 결과 ──
|
|
call_psearch_result(client, user_id, seq, args.raw)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|