6d2a706a48
왜: 체결이 없어도 마지막가는 유지인데, TTL로 None 만들고 매도/EOD를 건너뛰어 8/5 돌파·금요일 leftover가 남음. 호가필터 TTL 구멍과 같은 병. 넣을 파일 신규: kis_trader/engine/live_sell_price.py kis_trader/strategies/base.py (_ws_last_quote, _resolve_sell_price) 전략: momentum.py scalping.py tail_catch.py breakout.py range_break.py dart_strategy.py updow_strategy.py updown_feed.py us_momentum.py WS: ws_manager.py kis_ws.py kiwoom_ws.py ls_ws.py kis_ws_overseas.py kis_trader/web/live_config_schema.py (WS_PRICE_MAX_AGE_SEC 기본 0) database.py (키 주석 + legacy/ sys.path) kis_trader/execution/order_manager.py (잔고 있는데 40240000 ghost_purge 금지 — 같은 EOD 사고) EOD가 min_hold에 안 막히게 손본 momentum_hts_logic.py / scalping_engine.py / tail_engine.py (이 대화에서 손본 부분만 확인 후) 문서: docs/like_mcp.md/db_erd.md code_architecture.md (가격 TTL 문구) 메시지 초안 fix: 매수·매도 현재가를 TTL로 버리지 않음 (마지막 RAM) 횡보·체결 공백을 죽은 캐시로 오인해 None 처리하면 손절·EOD가 스킵된다. 호가필터와 같이 나이는 무시하고 마지막 체결가를 유지한다. EOD는 매수가 폴백. 영향: 실매 O / 백테·옵투나 봉 경로 거의 무관 (엔진 식 변경 아님) 커밋 2 — 루트 정리 (remove/ vs legacy/) 왜: 루트 단독봇·테스트는 지울 보관함으로. 웹·알람이 아직 쓰는 모듈은 remove에 두면 나중에 폴더째 삭제 때 깨짐. 넣을 파일 이동: 미사용 → remove/legacy_root/ (래퍼, ETF/키움 옛봇, 테스트, kiwoom_rest_api 등) 이동: 사용 중 → legacy/ (holding_bot kis_holding_ver1 news_analyzer kis_long_ver1/2) 신규: kis_trader/utils/legacy_root.py legacy/README.md remove/README.md import 경로: backtest_web.py mm_butler.py mm_remote.py updow_holding_cfg.py dbband_stock_cfg.py param_search_updow*.py dbband_param_search.py param_search_apply_snapshot.py verify_three_paths.py docs/like_mcp.md/code_architecture.md 수동 노트 메시지 초안 chore: 미사용 루트는 remove/, 웹·알람 구모듈은 legacy/ remove는 나중에 통째 삭제 예정. holding_bot·news_analyzer·kis_long은 ensure_legacy_root로 legacy/만 본다. 빼기: scratch/set_ws_price_max_age_zero.py (일회성)
999 lines
33 KiB
Python
999 lines
33 KiB
Python
#!/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())
|