kis_bot/remove/legacy_root/test_kiwoom_ws.py

#!/usr/bin/env python3
"""
test_kiwoom_ws.py — 키움 WebSocket 시세 단독 검증 스크립트
=========================================================

봇 안 띄우고 ``kiwoom_ws.KiwoomWebSocketPriceCache`` 만 직접 띄워서
연결·로그인·등록·시세 수신이 정상인지 30~60초 안에 확인.

사용법
------

기본 (DB 키 자동 로드, 삼성전자 + SK하이닉스 30초 구독)::

    python3 test_kiwoom_ws.py

종목 직접 지정::

    python3 test_kiwoom_ws.py --codes 005930,000660,005380 --duration 60

장외에 띄워도 LOGIN/REG ack 까지 확인 가능 (시세는 안 오지만 인증 동작 검증).

출력 해석
---------

✅ LOGIN OK         → 키움 토큰·키 정상
✅ REG OK           → 종목 등록 성공
📈 005930 73900...  → 1초 1줄, 실시간 가격 수신 정상
⚠️ 가격 수신 0건    → 장외 또는 키움 서버 정책
❌ LOGIN 실패       → 키 / OpenAPI 신청 / 도메인 문제 (test_kiwoom_token.py 로 추가 진단)
"""
from __future__ import annotations

import argparse
import logging
import sys
import time
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))


def setup_logging(verbose: bool) -> None:
    fmt = "[%(asctime)s] [%(name)s] %(message)s"
    logging.basicConfig(
        level=logging.DEBUG if verbose else logging.INFO,
        format=fmt,
        datefmt="%H:%M:%S",
    )


def load_kiwoom_creds(force_real: bool = True) -> tuple[str, str, bool]:
    """DB env_config 에서 키움 키 로드.

    Args:
        force_real: True (기본) 이면 KIS_MOCK 와 무관하게 항상 실키/실전 도메인.
                    시세 검증·실거래 시세 비교에서는 모의 도메인이 의미 없으므로 권장.
                    False 면 ``_get_kiwoom_creds()`` 가 KIS_MOCK 따라 자동 선택 (디버그용).
    """
    from database import TradeDB

    db = TradeDB()
    try:
        if force_real:
            row = db.conn.execute(
                "SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
            ).fetchone()
            if not row:
                return "", "", False
            r = dict(row)
            key = (r.get("KIWOOM_APP_KEY_REAL") or "").strip()
            secret = (r.get("KIWOOM_APP_SECRET_REAL") or "").strip()
            if not key or not secret:
                key = (r.get("KIWOOM_APP_KEY") or "").strip()
                secret = (r.get("KIWOOM_APP_SECRET") or "").strip()
            return key, secret, False  # 항상 실전 도메인
        from kis_trader.ws.kis_ws import _get_kiwoom_creds
        return _get_kiwoom_creds(db)
    finally:
        db.close()


def main() -> int:
    p = argparse.ArgumentParser(description="키움 WebSocket 시세 단독 검증")
    p.add_argument(
        "--codes",
        default="005930,000660",
        help="콤마 구분 종목코드 (기본: 삼성전자, SK하이닉스)",
    )
    p.add_argument(
        "--duration", type=int, default=30,
        help="구독 유지 시간(초). 기본 30",
    )
    p.add_argument("-v", "--verbose", action="store_true", help="DEBUG 로그")
    p.add_argument(
        "--mock-auto", action="store_true",
        help="KIS_MOCK 따라 자동 선택 (기본은 실키/실전 도메인 강제)",
    )
    args = p.parse_args()

    setup_logging(args.verbose)
    logger = logging.getLogger("test_kiwoom_ws")

    codes = [c.strip() for c in args.codes.split(",") if c.strip()]
    if not codes:
        print("❌ --codes 비어있음")
        return 2

    print("=" * 70)
    print("🔍 키움 WS 단독 검증")
    print("=" * 70)

    # 키 로드 (기본 실키 강제, --mock-auto 면 KIS_MOCK 따라 자동)
    force_real = not args.mock_auto
    try:
        app_key, app_secret, is_mock = load_kiwoom_creds(force_real=force_real)
    except Exception as e:
        print(f"❌ 키 로드 실패: {e}")
        return 1
    if not app_key or not app_secret:
        col = "KIWOOM_APP_KEY_REAL" if force_real else "KIWOOM_APP_KEY_*"
        print(f"❌ 키움 키 미설정 (DB env_config 의 {col} 확인)")
        return 1
    mode_label = "실전(force_real)" if force_real else f"auto (mock={is_mock})"
    print(f"  키움 키: {app_key[:8]}…{app_key[-4:]}  도메인={mode_label}")
    print(f"  대상 종목: {codes}")
    print(f"  유지 시간: {args.duration}초")
    print()

    # WS 인스턴스
    from kis_trader.ws.kiwoom_ws import KiwoomWebSocketPriceCache

    ws = KiwoomWebSocketPriceCache(app_key, app_secret, is_mock=is_mock)
    if not ws.start():
        print("❌ 키움 WS start() 실패")
        return 1

    # 종목 구독
    for code in codes:
        ws.subscribe(code)

    # 연결 + LOGIN 대기 (최대 15초)
    deadline = time.time() + 15
    while time.time() < deadline:
        if ws.is_connected():
            print(f"✅ LOGIN OK ({int(time.time() - (deadline - 15))}s)")
            break
        time.sleep(0.5)
    else:
        print("⚠️ LOGIN 미완료 (15초 timeout). 키움 토큰/네트워크 확인 필요")
        ws.stop()
        return 1

    # 가격 수신 모니터링
    print()
    print(f"📡 {args.duration}초간 시세 모니터링 (1초 간격 출력)...")
    end_ts = time.time() + args.duration
    last_dump_ts = 0.0
    received: dict = {}
    while time.time() < end_ts:
        now = time.time()
        if now - last_dump_ts >= 1.0:
            last_dump_ts = now
            for code in codes:
                d = ws.get_price(code, max_age_sec=10.0)
                if d:
                    px = d.get("stck_prpr", "?")
                    chg = d.get("prdy_ctrt", "?")
                    age = d.get("_age_ms", "?")
                    print(f"  📈 {code}  price={px}원  chg={chg}%  age={age}ms")
                    received[code] = received.get(code, 0) + 1
                else:
                    print(f"  ⏸  {code}  (no data — 장외/미수신)")
        time.sleep(0.2)

    # 종료
    print()
    print("─" * 70)
    print("종료. 수신 통계:")
    for code in codes:
        n = received.get(code, 0)
        mark = "✅" if n > 0 else "⚠️"
        print(f"  {mark} {code}: {n}회 수신")
    print("─" * 70)
    print()
    print("팁:")
    print("  • 모든 종목 0회면 장외 시간이거나 키움 정책 문제일 수 있음.")
    print("  • 장중에 0회면 → 키움 OpenAPI+ 신청 옵션 (실시간 시세 권한) 확인.")
    print("  • LOGIN 실패면 → test_kiwoom_token.py 로 토큰 발급 자체 진단.")

    ws.stop()
    return 0 if any(received.values()) else 1


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