kis_trader/remove/legacy_root/test_psearch.py

#!/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())