kis_bot/_test_kiwoom_condition_list.py

#!/usr/bin/env python3
"""
_test_kiwoom_condition_list.py — 키움 웹소켓 조건검색 '목록조회'(ka10171 / CNSRLST) 단독 테스트
==================================================================================================
[목적]
실매를 돌리지 않고, 키움 신형 REST/WS API 로 서버에 저장된 조건검색식 목록을
웹소켓으로 가져올 수 있는지만 확인한다. (유니버스 전환의 전제조건 검증)

[흐름]
1) DB(env_config)에서 키움 앱키/시크릿/모의여부 로드 (봇과 동일 키)
2) KiwoomTokenManager 로 access_token 발급 (au10001)
3) wss://api.kiwoom.com:10000/api/dostk/websocket 연결
4) {"trnm":"LOGIN","token":...} 인증
5) LOGIN OK → {"trnm":"CNSRLST"} 전송
6) CNSRLST 응답의 조건식 목록(seq, name) 출력 후 종료

[주의]
- 읽기 전용(조회)만 한다. 실시간 등록(CNSRREQ)·주문은 하지 않는다.
- 실제 응답 원문(raw)을 그대로 찍어, 필드명이 문서와 달라도 눈으로 확인 가능.
"""
import json
import os
import sys
import threading
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from kis_trader.utils.env import get_env_from_db
from kis_trader.ws.kis_ws import _get_kiwoom_token_cached


def _load_kiwoom_creds():
    """봇과 동일한 우선순위로 키움 키를 로드 (REAL → LEGACY, MOCK 토글)."""
    kw_mock_raw = (get_env_from_db("KIWOOM_MOCK", "") or "").strip().lower()
    is_mock = kw_mock_raw in ("1", "true", "y", "yes", "on")
    if is_mock:
        key = (get_env_from_db("KIWOOM_APP_KEY_MOCK", "") or "").strip()
        secret = (get_env_from_db("KIWOOM_APP_SECRET_MOCK", "") or "").strip()
    else:
        key = (get_env_from_db("KIWOOM_APP_KEY_REAL", "") or "").strip()
        secret = (get_env_from_db("KIWOOM_APP_SECRET_REAL", "") or "").strip()
    # 레거시 단일 필드 폴백
    if not key or not secret:
        key = key or (get_env_from_db("KIWOOM_APP_KEY", "") or "").strip()
        secret = secret or (get_env_from_db("KIWOOM_APP_SECRET", "") or "").strip()
    return key, secret, is_mock


def main() -> int:
    try:
        import websocket  # websocket-client
    except Exception as e:
        print(f"❌ websocket-client 미설치: {e}  (pip install websocket-client)")
        return 1

    key, secret, is_mock = _load_kiwoom_creds()
    if not key or not secret:
        print("❌ 키움 앱키/시크릿 미설정 — env_config 의 KIWOOM_APP_KEY_REAL/SECRET_REAL 확인")
        return 1
    print(f"🔑 키움 키 로드 OK (mock={is_mock}, key 앞8자={key[:8]}…)")

    token = _get_kiwoom_token_cached(key, secret, is_mock)
    if not token:
        print("❌ 키움 토큰 발급 실패 (au10001)")
        return 1
    print(f"🎫 토큰 발급 OK (앞8자={token[:8]}…)")

    url = (
        get_env_from_db("KIWOOM_WS_URL_MOCK", "wss://mockapi.kiwoom.com:10000/api/dostk/websocket")
        if is_mock else
        get_env_from_db("KIWOOM_WS_URL_REAL", "wss://api.kiwoom.com:10000/api/dostk/websocket")
    )
    print(f"🌐 WS 연결 시도: {url}")

    state = {"done": False, "conditions": None, "error": None}

    def on_open(ws):
        try:
            ws.send(json.dumps({"trnm": "LOGIN", "token": token}))
            print("📡 LOGIN 발송")
        except Exception as e:
            state["error"] = f"LOGIN 발송 실패: {e}"
            ws.close()

    def on_message(ws, message):
        try:
            data = json.loads(message)
        except Exception:
            print(f"📥 (raw, non-json) {message[:200]}")
            return
        trnm = data.get("trnm")

        # 서버 PING 은 그대로 echo (연결 유지)
        if trnm == "PING":
            try:
                ws.send(message)
            except Exception:
                pass
            return

        if trnm == "LOGIN":
            rc = str(data.get("return_code"))
            if rc in ("0", "0.0"):
                print("✅ LOGIN OK → CNSRLST(조건검색 목록조회) 요청")
                ws.send(json.dumps({"trnm": "CNSRLST"}))
            else:
                state["error"] = f"LOGIN 실패 rc={rc} msg={data.get('return_msg')}"
                ws.close()
            return

        if trnm == "CNSRLST":
            print("\n===== CNSRLST 응답 원문 =====")
            print(json.dumps(data, ensure_ascii=False, indent=2))
            state["conditions"] = data.get("data") or []
            state["done"] = True
            ws.close()
            return

        # 그 외 메시지도 원문 표시(디버깅용)
        print(f"📥 기타 메시지: {json.dumps(data, ensure_ascii=False)[:300]}")

    def on_error(ws, err):
        state["error"] = f"WS 오류: {err}"

    def on_close(ws, code, msg):
        state["done"] = True

    ws = websocket.WebSocketApp(
        url,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
    )
    th = threading.Thread(target=ws.run_forever, kwargs={"ping_interval": 0}, daemon=True)
    th.start()

    # 최대 20초 대기
    deadline = time.time() + 20
    while time.time() < deadline and not state["done"]:
        time.sleep(0.2)
    try:
        ws.close()
    except Exception:
        pass

    if state["error"]:
        print(f"\n🚨 {state['error']}")
        return 2

    conds = state["conditions"]
    if conds is None:
        print("\n⏱️  응답 없이 타임아웃 — 연결/인증 로그 확인 필요")
        return 3

    print(f"\n📋 조건검색식 {len(conds)}개:")
    momentum_seq = None
    for c in conds:
        # 응답 포맷: [seq, name] 배열 또는 {"seq":..,"name":..} dict 둘 다 허용
        if isinstance(c, (list, tuple)):
            seq, name = (c[0] if len(c) > 0 else ""), (c[1] if len(c) > 1 else "")
        else:
            seq, name = c.get("seq"), c.get("name")
        mark = ""
        if str(name).strip().lower() == "momentum":
            momentum_seq = seq
            mark = "  ← momentum"
        print(f"   seq={str(seq):>4}  name={name}{mark}")

    if momentum_seq is not None:
        print(f"\n🎯 'momentum' 조건식 발견: seq={momentum_seq}")
    else:
        print("\n⚠️  이름이 정확히 'momentum' 인 조건식은 목록에 없음(이름 확인 필요)")
    print("\n🎉 조건검색 목록조회(웹소켓) 성공 — 키움 유니버스 전환 전제조건 충족")
    return 0


if __name__ == "__main__":
    sys.exit(main())