#!/usr/bin/env python3 """ test_ls_ws.py — LS증권 OpenAPI WebSocket 실시간 구독 단독 검증 ============================================================= 봇과 무관하게 LS 접근토큰 발급 → (해외면 REST 현재가) → WS 구독 → **수신 body 필드 전체**를 한글로 덤프한다. 공식 참고 --------- - REST 토큰: ``POST https://openapi.ls-sec.co.kr:8080/oauth2/token`` - 실전 WS: ``wss://openapi.ls-sec.co.kr:9443/websocket`` - 모의 WS: ``wss://openapi.ls-sec.co.kr:29443/websocket`` - 해외 GSC ``tr_key``: ``{거래소코드}{심볼}`` 을 **18자리 공백 패딩** (예: NASDAQ TSLA → ``\"82TSLA\" + 공백12`` = 총 18자. 공식 예: ``81SOXL ``) - 해외 REST 현재가: ``POST /overseas-stock/market-data`` ``g3101`` 사용법 ------ 영구구독 KR:: python3 test_ls_ws.py 해외 SPCX / TSLA / QQQM (필드 덤프 + REST 현재가):: python3 test_ls_ws.py --codes SPCX,TSLA,QQQM --duration 45 """ from __future__ import annotations import argparse import json import logging import sys import threading import time from datetime import datetime from pathlib import Path from typing import Any import requests SCRIPT_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(SCRIPT_DIR)) LS_REST_BASE = "https://openapi.ls-sec.co.kr:8080" LS_WS_REAL = "wss://openapi.ls-sec.co.kr:9443/websocket" LS_WS_MOCK = "wss://openapi.ls-sec.co.kr:29443/websocket" KR_TR_UNIFIED = "US3" KR_TR_UNIFIED_HOGA = "UH1" KR_TR_KOSPI = "S3_" KR_TR_KOSDAQ = "K3_" KR_TR_KOSPI_HOGA = "H1_" KR_TR_KOSDAQ_HOGA = "HA_" US_TR_TRADE = "GSC" # LS/xing 해외 거래소코드 (미국) EXCHANGE_TO_LS: dict[str, str] = { "NASD": "82", "NASDAQ": "82", "NQ": "82", "NYSE": "81", "NYS": "81", "AMEX": "83", "AMS": "83", "ASE": "83", } # GSC 실시간 체결 OutBlock (스펙 blocks.json) GSC_FIELD_KO: dict[str, str] = { "symbol": "종목코드", "price": "현재가", "sign": "등락부호", "diff": "전일대비", "rate": "등락률(%)", "open": "시가", "high": "고가", "low": "저가", "high52p": "52주최고", "low52p": "52주최저", "trdq": "체결수량", "totq": "누적거래량", "amount": "누적거래대금", "cgubun": "체결구분", "trdtm": "체결시각(현지)", "kortm": "체결시각(한국)", "ovsdate": "현지일자", "kordate": "한국일자", "lSeq": "시퀀스", } # US3/S3_ 국내 체결 OutBlock (주요 필드) KR_TICK_FIELD_KO: dict[str, str] = { "shcode": "종목코드", "price": "현재가", "sign": "등락부호", "change": "전일대비", "drate": "등락률(%)", "open": "시가", "high": "고가", "low": "저가", "volume": "누적거래량", "cvolume": "체결량", "value": "누적거래대금", "chetime": "체결시각", "bidho": "매수호가", "offerho": "매도호가", "cpower": "체결강도", "w_avrg": "가중평균가", "exchname": "거래소", "status": "상태", } # g3101 REST 현재가 OutBlock G3101_FIELD_KO: dict[str, str] = { "symbol": "종목코드", "korname": "한글명", "exchcd": "거래소코드", "exchange": "거래소명", "currency": "통화", "price": "현재가", "sign": "등락부호", "diff": "전일대비", "rate": "등락률(%)", "open": "시가", "high": "고가", "low": "저가", "volume": "거래량", "amount": "거래대금", "high52p": "52주최고", "low52p": "52주최저", "uplimit": "상한가", "dnlimit": "하한가", "suspend": "거래정지", "sellonly": "매도만가능", "floatpoint": "소수점자리", "induname": "업종", "perv": "PER", "epsv": "EPS", "delaygb": "지연구분", "keysymbol": "키심볼", } def setup_logging(verbose: bool, log_path: Path | None) -> logging.Logger: logger = logging.getLogger("test_ls_ws") logger.handlers.clear() logger.setLevel(logging.DEBUG if verbose else logging.INFO) fmt = logging.Formatter("[%(asctime)s] %(message)s", datefmt="%H:%M:%S") sh = logging.StreamHandler(sys.stdout) sh.setFormatter(fmt) logger.addHandler(sh) if log_path is not None: log_path.parent.mkdir(parents=True, exist_ok=True) fh = logging.FileHandler(log_path, encoding="utf-8") fh.setFormatter(fmt) logger.addHandler(fh) return logger def load_ls_creds(*, use_mock: bool) -> tuple[str, str]: """DB env_config 에서 LS AppKey/Secret 로드.""" from database import TradeDB db = TradeDB() try: row = db.conn.execute( "SELECT LS_APP_KEY_REAL, LS_APP_SECRET_REAL, " "LS_APP_KEY_MOCK, LS_APP_SECRET_MOCK " "FROM env_config ORDER BY id DESC LIMIT 1" ).fetchone() if not row: return "", "" r = dict(row) if use_mock: key = (r.get("LS_APP_KEY_MOCK") or "").strip() secret = (r.get("LS_APP_SECRET_MOCK") or "").strip() else: key = (r.get("LS_APP_KEY_REAL") or "").strip() secret = (r.get("LS_APP_SECRET_REAL") or "").strip() return key, secret finally: db.close() def load_permanent_codes(market: str) -> list[dict[str, str]]: """permanent_subscriptions 에서 enabled=1 종목 로드.""" from database import TradeDB db = TradeDB() try: market_u = (market or "KR").strip().upper() if market_u == "ALL": sql = ( "SELECT code, market_type, exchange, symbol " "FROM permanent_subscriptions WHERE enabled=1 " "ORDER BY market_type, code" ) rows = db.conn.execute(sql).fetchall() else: sql = ( "SELECT code, market_type, exchange, symbol " "FROM permanent_subscriptions " "WHERE enabled=1 AND market_type=%s " "ORDER BY code" ) rows = db.conn.execute(sql, (market_u,)).fetchall() out: list[dict[str, str]] = [] for row in rows: r = dict(row) out.append( { "code": str(r.get("code") or "").strip(), "market_type": str(r.get("market_type") or "").strip().upper(), "exchange": str(r.get("exchange") or "").strip().upper(), "symbol": str(r.get("symbol") or r.get("code") or "").strip(), } ) return [x for x in out if x["code"]] finally: db.close() def fetch_access_token(app_key: str, app_secret: str, timeout: float = 15.0) -> str: """LS OAuth2 client_credentials → access_token.""" url = f"{LS_REST_BASE}/oauth2/token" headers = {"Content-Type": "application/x-www-form-urlencoded"} data = { "grant_type": "client_credentials", "appkey": app_key, "appsecretkey": app_secret, "scope": "oob", } resp = requests.post(url, headers=headers, data=data, timeout=timeout) if resp.status_code >= 400: raise RuntimeError(f"token HTTP {resp.status_code}: {resp.text[:400]}") body = resp.json() token = body.get("access_token") or body.get("accesstoken") if not token: raise RuntimeError(f"no access_token in response: {body}") return str(token) def ls_exchcd(exchange: str, default: str = "82") -> str: """영구구독 exchange(NASD/NYSE/AMEX) → LS 2자리 코드.""" ex = (exchange or "").strip().upper() if ex.isdigit() and len(ex) == 2: return ex return EXCHANGE_TO_LS.get(ex, default) def overseas_tr_key(exchcd: str, symbol: str, width: int = 18) -> str: """GSC tr_key = 거래소코드+심볼, 오른쪽 공백 패딩 **18자리**. LS API 가이드([해외주식] 실시간 시세 GSC): Length=18, 예) ``'82TSLA' + 공백 12자리`` / Request ``\"81SOXL \"`` """ raw = f"{exchcd}{symbol}" if len(raw) >= width: return raw return raw.ljust(width) def domestic_unified_tr_key(shcode: str, width: int = 10) -> str: """US3/UH1 tr_key = ``U`` + 6자리단축코드 + 공백3 = 총 10자리. 공식 예: ``\"U005930 \"`` """ code = (shcode or "").strip() if len(code) == 6 and code.isdigit(): raw = f"U{code}" else: raw = code if code.startswith("U") else f"U{code}" if len(raw) >= width: return raw return raw.ljust(width) def print_field_catalog(title: str, labels: dict[str, str]) -> None: print(f"\n📋 {title}") for k, ko in labels.items(): print(f" {k:12s} {ko}") def dump_body_fields( body: dict[str, Any], labels: dict[str, str], *, title: str = "", indent: str = " ", ) -> None: """수신 dict 를 한글 라벨 + 값으로 출력 (스펙에 없는 키도 전부).""" if title: print(title) if not isinstance(body, dict) or not body: print(f"{indent}(empty body)") return shown: set[str] = set() for key, ko in labels.items(): if key in body: print(f"{indent}{ko} ({key}): {body.get(key)!r}") shown.add(key) extras = [k for k in body.keys() if k not in shown] for key in extras: print(f"{indent}{key}: {body.get(key)!r}") def _ls_rest_headers( token: str, app_key: str, app_secret: str, tr_cd: str, ) -> dict[str, str]: """LS REST 공통 헤더. tr_cont 누락 시 IGW40010.""" return { "Content-Type": "application/json; charset=UTF-8", "authorization": f"Bearer {token}", "appkey": app_key, "appsecretkey": app_secret, "tr_cd": tr_cd, "tr_cont": "N", "tr_cont_key": "", } def fetch_overseas_quote_g3101( token: str, app_key: str, app_secret: str, *, exchcd: str, symbol: str, delaygb: str = "R", timeout: float = 15.0, ) -> dict[str, Any]: """해외주식 REST 현재가 (g3101). delaygb: R=실시간, D=지연.""" url = f"{LS_REST_BASE}/overseas-stock/market-data" keysymbol = f"{exchcd}{symbol}" headers = _ls_rest_headers(token, app_key, app_secret, "g3101") payload = { "g3101InBlock": { "delaygb": delaygb, "keysymbol": keysymbol, "exchcd": exchcd, "symbol": symbol, } } resp = requests.post(url, headers=headers, json=payload, timeout=timeout) try: data = resp.json() except Exception: return {"_http": resp.status_code, "_raw": resp.text[:500]} data["_http"] = resp.status_code return data def fetch_kr_quote_t1101( token: str, app_key: str, app_secret: str, *, shcode: str = "005930", timeout: float = 15.0, ) -> dict[str, Any]: """국내 주식현재가 (t1101) — 해외 API 빈응답일 때 필드 덤프 검증용.""" url = f"{LS_REST_BASE}/stock/market-data" headers = _ls_rest_headers(token, app_key, app_secret, "t1101") payload = {"t1101InBlock": {"shcode": shcode}} resp = requests.post(url, headers=headers, json=payload, timeout=timeout) try: data = resp.json() except Exception: return {"_http": resp.status_code, "_raw": resp.text[:500]} data["_http"] = resp.status_code return data # t1101 주요 필드 (호가 10단 전부는 생략, 핵심만 라벨) T1101_FIELD_KO: dict[str, str] = { "hname": "종목명", "price": "현재가", "sign": "등락부호", "change": "전일대비", "diff": "등락률(%)", "volume": "누적거래량", "jnilclose": "전일종가", "offerho1": "매도1호가", "bidho1": "매수1호가", "offerrem1": "매도1잔량", "bidrem1": "매수1잔량", "offer": "매도총잔량", "bid": "매수총잔량", "uplmtprice": "상한가", "dnlmtprice": "하한가", "open": "시가", "high": "고가", "low": "저가", "ho_yn": "관리종목", "shcode": "종목코드", "exchid": "거래소", } # H1_/UH1 호가 잔량 (1호가만 라벨, 나머지는 키 그대로 덤프) HOGA_FIELD_KO: dict[str, str] = { "shcode": "종목코드", "hotime": "호가시간", "offerho1": "매도1호가", "bidho1": "매수1호가", "offerrem1": "매도1잔량", "bidrem1": "매수1잔량", "totofferrem": "총매도잔량", "totbidrem": "총매수잔량", "volume": "누적거래량", "donsigubun": "동시호가구분", "midprice": "중간가격", } def build_subscriptions( items: list[dict[str, str]], *, kr_tr: str, also_hoga: bool = False, ) -> list[tuple[str, str, str]]: """(tr_cd, tr_key, label) 목록. 국내 통합(US3/UH1): tr_key = ``U005930 `` (U+6자리+공백3) 국내 단독(S3_/H1_/K3_/HA_): tr_key = ``005930`` (6자리) """ subs: list[tuple[str, str, str]] = [] seen: set[tuple[str, str]] = set() def add(tr_cd: str, tr_key: str, label: str) -> None: key = (tr_cd, tr_key) if key in seen: return seen.add(key) subs.append((tr_cd, tr_key, label)) mode = (kr_tr or "unified").strip().lower() for it in items: code = it["code"] mt = it["market_type"] if mt == "US": sym = it.get("symbol") or code exchcd = ls_exchcd(it.get("exchange") or "NASD") tr_key = overseas_tr_key(exchcd, sym) add(US_TR_TRADE, tr_key, f"US:{sym} exch={exchcd}") continue # KR if mode == "dual": add(KR_TR_KOSPI, code, f"KR:{code}:S3_") add(KR_TR_KOSDAQ, code, f"KR:{code}:K3_") if also_hoga: add(KR_TR_KOSPI_HOGA, code, f"KR:{code}:H1_") add(KR_TR_KOSDAQ_HOGA, code, f"KR:{code}:HA_") elif mode == "kospi": add(KR_TR_KOSPI, code, f"KR:{code}:S3_") if also_hoga: add(KR_TR_KOSPI_HOGA, code, f"KR:{code}:H1_") elif mode == "kosdaq": add(KR_TR_KOSDAQ, code, f"KR:{code}:K3_") if also_hoga: add(KR_TR_KOSDAQ_HOGA, code, f"KR:{code}:HA_") else: ukey = domestic_unified_tr_key(code) add(KR_TR_UNIFIED, ukey, f"KR:{code}:US3") if also_hoga: add(KR_TR_UNIFIED_HOGA, ukey, f"KR:{code}:UH1") return subs def _is_tick_body(body: Any) -> bool: if not isinstance(body, dict) or not body: return False tick_keys = ( "price", "chetime", "drate", "cvolume", "trdq", "totq", "rate", "kortm", "trdtm", "shcode", "symbol", "offerho1", "bidho1", "offerrem1", "bidrem1", ) return any(k in body for k in tick_keys) class LsWsProbe: """websocket-client 기반 짧은 구독 프로브.""" def __init__( self, ws_url: str, token: str, logger: logging.Logger, *, verbose: bool = False, max_full_dumps: int = 3, ) -> None: try: import websocket # websocket-client except ImportError as e: raise RuntimeError( "websocket-client 필요: pip install websocket-client" ) from e self._websocket = websocket self.ws_url = ws_url self.token = token self.logger = logger self.verbose = verbose self.max_full_dumps = max_full_dumps self._ws: Any = None self._thread: threading.Thread | None = None self.opened = threading.Event() self._stop = threading.Event() self.msg_count = 0 self.tick_count = 0 self.reg_acks = 0 self.errors: list[str] = [] self.by_label: dict[str, int] = {} self._label_by_key: dict[tuple[str, str], str] = {} self._label_by_key_stripped: dict[tuple[str, str], str] = {} self._last_tick: dict[str, Any] = {} self._full_dumps_left = max_full_dumps self._lock = threading.Lock() def start(self) -> None: self._ws = self._websocket.WebSocketApp( self.ws_url, on_open=self._on_open, on_message=self._on_message, on_error=self._on_error, on_close=self._on_close, ) self._thread = threading.Thread( target=self._run, name="ls-ws-probe", daemon=True ) self._thread.start() def _run(self) -> None: assert self._ws is not None self._ws.run_forever(ping_interval=20, ping_timeout=10) def wait_open(self, timeout: float = 15.0) -> bool: return self.opened.wait(timeout=timeout) def register(self, tr_cd: str, tr_key: str, label: str) -> None: self._label_by_key[(tr_cd, tr_key)] = label self._label_by_key_stripped[(tr_cd, tr_key.strip())] = label payload = { "header": {"token": self.token, "tr_type": "3"}, "body": {"tr_cd": tr_cd, "tr_key": tr_key}, } if self._ws is None: raise RuntimeError("WS not started") self._ws.send(json.dumps(payload, ensure_ascii=False)) self.logger.info( "REG sent tr_cd=%s tr_key=%r (%s)", tr_cd, tr_key, label ) def stop(self) -> None: self._stop.set() if self._ws is not None: try: self._ws.close() except Exception: pass def _resolve_label(self, tr_cd: str, tr_key: str) -> str: if (tr_cd, tr_key) in self._label_by_key: return self._label_by_key[(tr_cd, tr_key)] stripped = tr_key.strip() if (tr_cd, stripped) in self._label_by_key_stripped: return self._label_by_key_stripped[(tr_cd, stripped)] # symbol only match (GSC body.symbol) for (tcd, tk), lab in self._label_by_key.items(): if tcd == tr_cd and (stripped in tk or tk.strip().endswith(stripped)): return lab return f"{tr_cd}:{tr_key!r}" def _on_open(self, _ws: Any) -> None: self.logger.info("WS OPEN %s", self.ws_url) self.opened.set() def _on_close(self, _ws: Any, status: Any, msg: Any) -> None: self.logger.info("WS CLOSE status=%s msg=%s", status, msg) def _on_error(self, _ws: Any, err: Any) -> None: text = str(err) self.errors.append(text) self.logger.error("WS ERROR %s", text) def _on_message(self, _ws: Any, message: Any) -> None: self.msg_count += 1 try: data = json.loads(message) if isinstance(message, str) else message except Exception: self.logger.warning("non-JSON msg: %s", str(message)[:200]) return header = data.get("header") or {} body = data.get("body") or {} tr_cd = str(header.get("tr_cd") or body.get("tr_cd") or "") tr_key = str( header.get("tr_key") or body.get("tr_key") or body.get("shcode") or body.get("symbol") or "" ) rsp_cd = str(header.get("rsp_cd") or "") rsp_msg = str(header.get("rsp_msg") or "") if not _is_tick_body(body): if rsp_msg or rsp_cd: self.reg_acks += 1 self.logger.info( "REG ack tr_cd=%s tr_key=%r rsp=%s %s", tr_cd, tr_key, rsp_cd, rsp_msg, ) return label = self._resolve_label(tr_cd, tr_key) with self._lock: self.tick_count += 1 self.by_label[label] = self.by_label.get(label, 0) + 1 self._last_tick[label] = dict(body) if isinstance(body, dict) else body do_full = self._full_dumps_left > 0 if do_full: self._full_dumps_left -= 1 labels = GSC_FIELD_KO if tr_cd == "GSC" else KR_TICK_FIELD_KO if tr_cd in ("UH1", "H1_", "HA_", "NH1", "B7_"): labels = HOGA_FIELD_KO px = body.get("price", body.get("offerho1", "?")) rate = body.get("rate", body.get("drate", "?")) tmk = ( body.get("kortm") or body.get("trdtm") or body.get("chetime") or body.get("hotime") or "" ) print( f"\n📈 TICK #{self.tick_count} {label} " f"price/ask={px} rate={rate} time={tmk} tr={tr_cd}" ) if do_full or self.verbose: dump_body_fields(body, labels, title=" [전문 body 전체]") raw = json.dumps(body, ensure_ascii=False) print(f" [raw JSON] {raw[:900]}{'…' if len(raw) > 900 else ''}") else: # 요약만 (이미 전체 dump 한도 초과) for k in ( "symbol", "shcode", "price", "diff", "rate", "drate", "trdq", "cvolume", "totq", "volume", "offerho1", "bidho1", "offerrem1", "bidrem1", ): if k in body: ko = labels.get(k, k) print(f" {ko} ({k}): {body[k]!r}") def main() -> int: p = argparse.ArgumentParser(description="LS증권 WebSocket 실시간 구독 단독 검증") p.add_argument("--codes", default="", help="콤마 구분 종목. 예: SPCX,TSLA,QQQM") p.add_argument( "--market", default="KR", choices=["KR", "US", "all", "ALL"], help="영구구독 market_type (기본 KR). --codes 있으면 무시", ) p.add_argument( "--kr-tr", default="unified", choices=["unified", "dual", "kospi", "kosdaq"], help="국내 TR (기본 US3 통합). dual=S3_+K3_", ) p.add_argument( "--also-hoga", action="store_true", help="국내 호가잔량도 구독 (unified→UH1, kospi→H1_, kosdaq→HA_)", ) p.add_argument("--duration", type=int, default=30, help="구독 유지 초") p.add_argument("--mock", action="store_true", help="모의 키 + WS 29443") p.add_argument( "--no-rest-quote", action="store_true", help="해외 g3101 / 국내 t1101 REST 현재가 조회 생략", ) p.add_argument( "--delaygb", default="R", help="g3101 delaygb (R=실시간 기본, D=지연)", ) p.add_argument( "--dump-n", type=int, default=5, help="WS tick 전문 전체 덤프 횟수 (기본 5)", ) p.add_argument("-v", "--verbose", action="store_true") p.add_argument("--log", default="") args = p.parse_args() ts = datetime.now().strftime("%Y%m%d_%H%M%S") log_path = Path(args.log) if args.log else (SCRIPT_DIR / "logs" / f"ls_ws_{ts}.log") logger = setup_logging(args.verbose, log_path) print("=" * 70) print("🔍 LS증권 WS 실시간 구독 검증 (필드 덤프)") print("=" * 70) use_mock = bool(args.mock) try: app_key, app_secret = load_ls_creds(use_mock=use_mock) except Exception as e: print(f"❌ LS 키 로드 실패: {e}") return 1 if not app_key or not app_secret: cols = ( "LS_APP_KEY_MOCK / LS_APP_SECRET_MOCK" if use_mock else "LS_APP_KEY_REAL / LS_APP_SECRET_REAL" ) print(f"❌ LS 키 비어있음 (env_config 의 {cols})") return 1 ws_url = LS_WS_MOCK if use_mock else LS_WS_REAL print(f" 모드: {'모의' if use_mock else '실전'}") print(f" AppKey: {app_key[:8]}…{app_key[-4:]}") print(f" WS: {ws_url}") print(f" 로그: {log_path}") if args.codes.strip(): items: list[dict[str, str]] = [] for raw in args.codes.split(","): c = raw.strip() if not c: continue mt = "KR" if (c.isdigit() and len(c) == 6) else "US" # 미국 티커 기본 NASDAQ(82). AMEX 등은 --codes 에 EX:SYM 형태로 가능 exchange = "NASD" symbol = c if mt == "US" and ":" in c: exchange, symbol = c.split(":", 1) exchange, symbol = exchange.strip().upper(), symbol.strip().upper() c = symbol items.append( { "code": c, "market_type": mt, "exchange": exchange if mt == "US" else "", "symbol": symbol if mt == "US" else c, } ) print(f" 종목: --codes ({len(items)}개)") else: mkt = "ALL" if str(args.market).upper() == "ALL" else str(args.market).upper() try: items = load_permanent_codes(mkt) except Exception as e: print(f"❌ permanent_subscriptions 로드 실패: {e}") return 1 print(f" 종목: permanent_subscriptions enabled ({mkt}) {len(items)}개") if not items: print("❌ 구독 대상 종목 없음") return 2 for it in items: print( f" - {it['market_type']} {it['code']}" + (f" ({it['exchange']})" if it.get("exchange") else "") ) us_items = [x for x in items if x["market_type"] == "US"] if us_items: print_field_catalog("GSC(해외 실시간 체결) 수신 필드", GSC_FIELD_KO) print_field_catalog("g3101(REST 현재가) 수신 필드", G3101_FIELD_KO) if any(x["market_type"] == "KR" for x in items): print_field_catalog("US3/S3_(국내 체결) 주요 수신 필드", KR_TICK_FIELD_KO) if args.also_hoga: print_field_catalog("UH1/H1_(호가잔량) 주요 수신 필드", HOGA_FIELD_KO) print( " ※ LS [주식] 실시간 시세 TR 약 65종 " "(체결/호가/거래원/프로그램/VI/NXT·통합…)\n" " → 키움·한투에서 REST로 떼오던 호가·거래원 등을 WS로 바로 구독 가능" ) subs = build_subscriptions( items, kr_tr=args.kr_tr, also_hoga=bool(args.also_hoga) ) print(f"\n 구독 슬롯: {len(subs)} (kr-tr={args.kr_tr})") for tr_cd, tr_key, label in subs: print(f" · {label} tr_cd={tr_cd} tr_key={tr_key!r}") print(f" 유지: {args.duration}초") print() try: token = fetch_access_token(app_key, app_secret) except Exception as e: print(f"❌ TOKEN 실패: {e}") logger.error("TOKEN fail: %s", e) return 1 print(f"✅ TOKEN OK ({token[:12]}…)") logger.info("TOKEN OK len=%d", len(token)) # ── REST 현재가 (해외) — WS tick 없어도 값 확인 ───────────────────── us_quote_ok = 0 if us_items and not args.no_rest_quote: print("\n" + "═" * 70) print(f"💵 REST g3101 현재가 (delaygb={args.delaygb})") print("═" * 70) for it in us_items: sym = it.get("symbol") or it["code"] exchcd = ls_exchcd(it.get("exchange") or "NASD") print(f"\n── {sym} exchcd={exchcd} keysymbol={exchcd}{sym}") try: data = fetch_overseas_quote_g3101( token, app_key, app_secret, exchcd=exchcd, symbol=sym, delaygb=str(args.delaygb), ) except Exception as e: print(f" ❌ REST 실패: {e}") continue rsp_cd = str(data.get("rsp_cd") or "") rsp_msg = str(data.get("rsp_msg") or "") print(f" http={data.get('_http')} rsp={rsp_cd!r} msg={rsp_msg!r}") out = data.get("g3101OutBlock") if isinstance(out, dict) and out: us_quote_ok += 1 dump_body_fields(out, G3101_FIELD_KO, title=" [g3101OutBlock — 실제 수신값]") print(f" [raw] {json.dumps(out, ensure_ascii=False)}") else: print(" ⚠️ g3101OutBlock 없음 (해외주식 API 미신청/권한 또는 종목키 문제)") print(f" [raw] {json.dumps(data, ensure_ascii=False)[:600]}") time.sleep(0.15) # TPS 여유 if us_quote_ok == 0: print("\n" + "─" * 70) print( "ℹ️ 해외 g3101 이 비어 있음.\n" " LS 홈페이지에서 「해외주식 API」 사용신청이 별도인지 확인.\n" " (국내 주식 API만 된 키면 g3101/GSC 값이 안 옴)\n" " 아래는 같은 키로 국내 t1101 을 조회한 **실제 필드 덤프 샘플**." ) print("─" * 70) try: kr = fetch_kr_quote_t1101( token, app_key, app_secret, shcode="005930" ) out = kr.get("t1101OutBlock") or {} print( f"\n── 국내 샘플 005930 삼성전자 " f"http={kr.get('_http')} rsp={kr.get('rsp_cd')!r}" ) if isinstance(out, dict) and out: # 핵심 필드만 + 일부 호가 dump_body_fields(out, T1101_FIELD_KO, title=" [t1101OutBlock 핵심]") # 너무 긴 호가10단은 raw 앞부분만 raw = json.dumps(out, ensure_ascii=False) print(f" [raw 앞 700자] {raw[:700]}…") else: print(f" {json.dumps(kr, ensure_ascii=False)[:500]}") except Exception as e: print(f" 국내 샘플 실패: {e}") # ── REST 현재가 (국내 t1101) — 장외에도 스냅샷 값 확인 ─────────────── kr_items = [x for x in items if x["market_type"] == "KR"] if kr_items and not args.no_rest_quote: print("\n" + "═" * 70) print("💵 REST t1101 국내 현재가/호가 (영구구독 KR)") print("═" * 70) for it in kr_items: code = it["code"] print(f"\n── {code}") try: data = fetch_kr_quote_t1101( token, app_key, app_secret, shcode=code ) except Exception as e: print(f" ❌ REST 실패: {e}") continue out = data.get("t1101OutBlock") or {} print( f" http={data.get('_http')} " f"rsp={data.get('rsp_cd')!r} msg={data.get('rsp_msg')!r}" ) if isinstance(out, dict) and out: dump_body_fields(out, T1101_FIELD_KO, title=" [t1101OutBlock 핵심]") else: print(f" [raw] {json.dumps(data, ensure_ascii=False)[:500]}") time.sleep(0.12) # ── WS ───────────────────────────────────────────────────────────── probe = LsWsProbe( ws_url, token, logger, verbose=args.verbose, max_full_dumps=max(1, int(args.dump_n)), ) probe.start() if not probe.wait_open(timeout=15.0): print("❌ WS OPEN timeout (15s)") probe.stop() return 1 print("\n✅ WS OPEN") for tr_cd, tr_key, label in subs: try: probe.register(tr_cd, tr_key, label) time.sleep(0.05) except Exception as e: print(f"❌ REG 실패 {label}: {e}") logger.error("REG fail %s: %s", label, e) print() print(f"📡 {args.duration}초간 WS 수신 (전문 필드 덤프 최대 {args.dump_n}건)...") end_ts = time.time() + max(1, int(args.duration)) last_summary = 0.0 while time.time() < end_ts: now = time.time() if now - last_summary >= 5.0: last_summary = now print( f" … msg={probe.msg_count} tick={probe.tick_count} " f"ack={probe.reg_acks} err={len(probe.errors)}" ) time.sleep(0.2) probe.stop() time.sleep(0.3) print() print("─" * 70) print("종료 통계") print(f" 메시지 총합: {probe.msg_count}") print(f" REG ack: {probe.reg_acks}") print(f" TICK: {probe.tick_count}") print(f" WS 에러: {len(probe.errors)}") if probe._last_tick: print("\n 마지막 tick 스냅샷 (종목별):") for label, body in sorted(probe._last_tick.items()): labels = GSC_FIELD_KO if "US:" in label else KR_TICK_FIELD_KO print(f"\n ▸ {label}") if isinstance(body, dict): dump_body_fields(body, labels) else: print(f" {body!r}") elif us_items: print( " ⚠️ WS tick 0 — 해외 실시간(GSC) 미수신.\n" " 위 REST g3101 값이 있으면 조회 API는 정상.\n" " tick 없으면 HTS [2007] 해외주식 실시간 시세신청 필요할 수 있음." ) else: print(" ⚠️ tick 0 — 장외이거나 권한/TR 확인") print(f"\n 로그: {log_path}") print("─" * 70) if probe.errors and probe.msg_count == 0: return 1 return 0 if __name__ == "__main__": sys.exit(main())