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 (일회성)
454 lines
16 KiB
Python
454 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_kiwoom_ws_sub_limit.py — 키움 WS 구독 한도(~100) 실측 + 로테이션
|
|
====================================================================
|
|
|
|
목적
|
|
----
|
|
1) 클라이언트 한도를 잠시 풀어 **120종목 REG** 를 보내 서버 반응 확인
|
|
2) 한도 초과 시 **REMOVE → REG** 로테이션이 되는지 확인
|
|
(국장/미장 시간대가 달라도, 구독 슬롯을 비우고 다시 넣는 패턴)
|
|
|
|
주의 (필수 읽기)
|
|
----------------
|
|
- 키움은 **앱키/토큰당 WS 1접속** 경향. 실매 ``kis_trader_main`` 이
|
|
``WS_PROVIDER=kis_with_validation`` 이면 키움 WS 를 이미 쓰고 있음.
|
|
- 이 테스트가 붙으면 **실매 키움 세션이 끊길 수 있음** (보통 재접속).
|
|
- 기본은 실매 감지 시 **중단**. 강제로 돌리려면 ``--force``.
|
|
|
|
사용법
|
|
------
|
|
::
|
|
|
|
# 안전: 실매 키움 사용 중이면 중단
|
|
python3 -u test_kiwoom_ws_sub_limit.py --count 120
|
|
|
|
# 실매 떠 있어도 강제 (장외·검증용)
|
|
nohup python3 -u test_kiwoom_ws_sub_limit.py --count 120 --force \\
|
|
--rotate-every 20 --rotate-n 25 --duration 90 \\
|
|
> logs/kiwoom_ws_sub_limit_$(date +%Y%m%d_%H%M%S).log 2>&1 &
|
|
|
|
로그: ``logs/kiwoom_ws_sub_limit_*.log`` 를 ``tail -f``
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections import deque
|
|
from pathlib import Path
|
|
from typing import Deque, List, Optional, Set
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
LOG_DIR = SCRIPT_DIR / "logs"
|
|
|
|
|
|
def setup_logging(log_path: Optional[Path], verbose: bool) -> logging.Logger:
|
|
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
|
logger = logging.getLogger("kiwoom_sub_limit")
|
|
logger.handlers.clear()
|
|
logger.setLevel(logging.DEBUG if verbose else logging.INFO)
|
|
fmt = logging.Formatter("[%(asctime)s] %(levelname)s %(message)s", "%H:%M:%S")
|
|
sh = logging.StreamHandler(sys.stdout)
|
|
sh.setFormatter(fmt)
|
|
logger.addHandler(sh)
|
|
if log_path:
|
|
fh = logging.FileHandler(log_path, encoding="utf-8")
|
|
fh.setFormatter(fmt)
|
|
logger.addHandler(fh)
|
|
# 키움 WS 내부 로그도 같은 핸들로
|
|
kw = logging.getLogger("KiwoomWebSocket")
|
|
kw.setLevel(logging.INFO)
|
|
kw.handlers.clear()
|
|
kw.addHandler(sh)
|
|
if log_path:
|
|
kw.addHandler(fh)
|
|
kw.propagate = False
|
|
return logger
|
|
|
|
|
|
def live_bot_may_hold_kiwoom() -> bool:
|
|
"""실매 유닛이 떠 있고 WS_PROVIDER 가 키움을 쓰는 설정이면 True."""
|
|
try:
|
|
out = subprocess.check_output(
|
|
["systemctl", "is-active", "kis_trader_main.service"],
|
|
text=True,
|
|
stderr=subprocess.DEVNULL,
|
|
).strip()
|
|
if out != "active":
|
|
return False
|
|
except Exception:
|
|
return False
|
|
try:
|
|
from kis_trader.utils.env import get_env_from_db
|
|
|
|
prov = (get_env_from_db("WS_PROVIDER", "") or "").strip().lower()
|
|
# kis_only 만 키움 미사용. 그 외(kis_with_validation / kiwoom_*) 는 위험.
|
|
return prov != "kis_only"
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
def load_kiwoom_creds(force_real: bool = True) -> tuple[str, str, bool]:
|
|
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 load_codes(count: int) -> List[str]:
|
|
"""stock_share_meta → ws_ticks → 하드코드 순으로 6자리 종목코드 수집."""
|
|
codes: List[str] = []
|
|
seen: Set[str] = set()
|
|
|
|
def _add(raw: str) -> None:
|
|
c = str(raw or "").strip()
|
|
if not re.fullmatch(r"\d{6}", c):
|
|
return
|
|
if c in seen:
|
|
return
|
|
seen.add(c)
|
|
codes.append(c)
|
|
|
|
from database import TradeDB
|
|
|
|
db = TradeDB()
|
|
try:
|
|
rows = db.conn.execute(
|
|
"SELECT code FROM stock_share_meta ORDER BY updated_at DESC LIMIT %s",
|
|
(max(count * 2, 300),),
|
|
).fetchall()
|
|
for r in rows:
|
|
_add(r["code"] if isinstance(r, dict) else r[0])
|
|
if len(codes) >= count:
|
|
break
|
|
if len(codes) < count:
|
|
rows2 = db.conn.execute(
|
|
"SELECT DISTINCT code FROM ws_ticks ORDER BY code LIMIT %s",
|
|
(count,),
|
|
).fetchall()
|
|
for r in rows2:
|
|
_add(r["code"] if isinstance(r, dict) else r[0])
|
|
if len(codes) >= count:
|
|
break
|
|
finally:
|
|
db.close()
|
|
|
|
# 그래도 부족하면 대표 대형주 패딩 (한도 테스트용 — 시세 품질보다 개수)
|
|
pad = [
|
|
"005930", "000660", "005380", "035420", "035720", "051910", "006400",
|
|
"005490", "028260", "105560", "055550", "012330", "032830", "066570",
|
|
"003550", "017670", "030200", "086790", "316140", "024110", "009150",
|
|
"034730", "018260", "011200", "010130", "009540", "010950", "011070",
|
|
"000270", "207940", "068270", "373220", "259960", "352820", "247540",
|
|
]
|
|
for c in pad:
|
|
_add(c)
|
|
if len(codes) >= count:
|
|
break
|
|
|
|
return codes[:count]
|
|
|
|
|
|
def patch_ws_for_limit_test(ws, client_cap: int) -> dict:
|
|
"""
|
|
- 클라이언트 한도를 client_cap 으로 올려 서버 한도(≈100)를 넘기게 함
|
|
- 0B 만 REG (호가/프로그램 제외 → 메시지·유량 절약)
|
|
- REG ack 카운터 훅
|
|
"""
|
|
stats = {
|
|
"reg_ok": 0,
|
|
"reg_fail": 0,
|
|
"reg_fail_msgs": [],
|
|
"login_ok": 0,
|
|
"login_fail": 0,
|
|
"real_ticks": 0,
|
|
}
|
|
|
|
ws._max_subscriptions = lambda: int(client_cap) # type: ignore[method-assign]
|
|
ws._orderbook_ws_enabled = lambda: False # type: ignore[method-assign]
|
|
ws._program_ws_enabled = lambda: False # type: ignore[method-assign]
|
|
ws._reg_types = lambda: [ws.SUB_TYPE] # type: ignore[method-assign]
|
|
|
|
orig_on_message = ws._on_message
|
|
|
|
def _wrapped(sock, message: str):
|
|
try:
|
|
import json
|
|
|
|
msg = json.loads(message)
|
|
trnm = msg.get("trnm", "")
|
|
if trnm == "LOGIN":
|
|
if msg.get("return_code") == 0:
|
|
stats["login_ok"] += 1
|
|
else:
|
|
stats["login_fail"] += 1
|
|
stats["reg_fail_msgs"].append(
|
|
f"LOGIN fail: {msg.get('return_msg')}"
|
|
)
|
|
elif trnm == "REG":
|
|
if msg.get("return_code") == 0:
|
|
stats["reg_ok"] += 1
|
|
else:
|
|
stats["reg_fail"] += 1
|
|
rm = str(msg.get("return_msg") or msg)
|
|
stats["reg_fail_msgs"].append(rm[:200])
|
|
elif trnm == "REAL":
|
|
stats["real_ticks"] += 1
|
|
except Exception:
|
|
pass
|
|
return orig_on_message(sock, message)
|
|
|
|
ws._on_message = _wrapped # type: ignore[method-assign]
|
|
return stats
|
|
|
|
|
|
def wait_login(ws, timeout: float, log: logging.Logger) -> bool:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if ws.is_authenticated():
|
|
log.info("✅ LOGIN OK")
|
|
return True
|
|
time.sleep(0.3)
|
|
log.error("❌ LOGIN timeout %.0fs", timeout)
|
|
return False
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="키움 WS 구독 한도·로테이션 실측")
|
|
p.add_argument("--count", type=int, default=120, help="시도 구독 종목 수 (기본 120)")
|
|
p.add_argument(
|
|
"--client-cap",
|
|
type=int,
|
|
default=200,
|
|
help="클라이언트 내부 한도 일시 상향 (서버 한도 실측용, 기본 200)",
|
|
)
|
|
p.add_argument(
|
|
"--soft-cap",
|
|
type=int,
|
|
default=100,
|
|
help="로테이션 시 유지할 목표 구독 수 (기본 100)",
|
|
)
|
|
p.add_argument("--duration", type=int, default=90, help="전체 테스트 초 (기본 90)")
|
|
p.add_argument(
|
|
"--rotate-every",
|
|
type=int,
|
|
default=20,
|
|
help="로테이션 주기(초). 0이면 로테이션 없음",
|
|
)
|
|
p.add_argument(
|
|
"--rotate-n",
|
|
type=int,
|
|
default=25,
|
|
help="한 번에 빼는/넣는 종목 수",
|
|
)
|
|
p.add_argument("--force", action="store_true", help="실매 키움 세션 충돌 무시")
|
|
p.add_argument("-v", "--verbose", action="store_true")
|
|
args = p.parse_args()
|
|
|
|
ts = time.strftime("%Y%m%d_%H%M%S")
|
|
log_path = LOG_DIR / f"kiwoom_ws_sub_limit_{ts}.log"
|
|
log = setup_logging(log_path, args.verbose)
|
|
|
|
log.info("=" * 70)
|
|
log.info("키움 WS 구독 한도 실측 count=%d soft_cap=%d client_cap=%d",
|
|
args.count, args.soft_cap, args.client_cap)
|
|
log.info("로그: %s", log_path)
|
|
log.info("=" * 70)
|
|
|
|
if live_bot_may_hold_kiwoom() and not args.force:
|
|
log.error(
|
|
"실매 kis_trader_main 이 키움 WS 를 쓸 수 있는 설정입니다. "
|
|
"이 테스트는 토큰당 1접속이라 실매 키움 세션을 끊을 수 있습니다. "
|
|
"장외·의도적 검증이면 --force 를 붙이세요."
|
|
)
|
|
return 2
|
|
|
|
if live_bot_may_hold_kiwoom() and args.force:
|
|
log.warning("⚠️ --force: 실매 키움 WS 가 끊길 수 있음 (보통 재접속)")
|
|
|
|
key, secret, is_mock = load_kiwoom_creds(force_real=True)
|
|
if not key or not secret:
|
|
log.error("키움 실키 없음 (KIWOOM_APP_KEY_REAL)")
|
|
return 1
|
|
log.info("키: %s…%s mock=%s", key[:8], key[-4:], is_mock)
|
|
|
|
all_codes = load_codes(args.count)
|
|
if len(all_codes) < args.count:
|
|
log.warning("종목 부족 %d/%d — 있는 만큼만", len(all_codes), args.count)
|
|
if len(all_codes) < 10:
|
|
log.error("종목 너무 적음")
|
|
return 1
|
|
log.info("종목 풀 %d개 (앞5: %s)", len(all_codes), all_codes[:5])
|
|
|
|
from kis_trader.ws.kiwoom_ws import KiwoomWebSocketPriceCache
|
|
|
|
ws = KiwoomWebSocketPriceCache(key, secret, is_mock=is_mock)
|
|
stats = patch_ws_for_limit_test(ws, client_cap=args.client_cap)
|
|
|
|
if not ws.start():
|
|
log.error("WS start 실패")
|
|
return 1
|
|
|
|
# Phase 1: 전부 넣기 (클라이언트는 client_cap 까지 허용 → 서버가 거절하는지 봄)
|
|
log.info("── Phase1: REG 시도 %d종목 (chunked) ──", len(all_codes))
|
|
added = ws.subscribe_many(all_codes)
|
|
log.info("클라이언트 집합 추가: %d / 요청 %d", len(added), len(all_codes))
|
|
|
|
if not wait_login(ws, 20.0, log):
|
|
ws.stop()
|
|
return 1
|
|
|
|
# REG 청크 전송 대기 (120 / 25 * 0.18 ≈ 1초 + 여유)
|
|
wait_reg = max(5.0, (len(all_codes) / 25.0) * 0.25 + 3.0)
|
|
log.info("REG 전송·ack 대기 %.1fs …", wait_reg)
|
|
time.sleep(wait_reg)
|
|
|
|
with ws._sub_lock:
|
|
n_sub = len(ws._subscribed)
|
|
log.info(
|
|
"Phase1 결과: subscribed=%d reg_ok=%d reg_fail=%d real_ticks=%d",
|
|
n_sub, stats["reg_ok"], stats["reg_fail"], stats["real_ticks"],
|
|
)
|
|
for m in stats["reg_fail_msgs"][:8]:
|
|
log.warning(" REG/LOGIN msg: %s", m)
|
|
|
|
# Phase 2: soft_cap 초과분이 있으면 로테이션 풀로 이동
|
|
soft = max(1, int(args.soft_cap))
|
|
rotate_n = max(1, int(args.rotate_n))
|
|
active: Deque[str] = deque()
|
|
overflow: Deque[str] = deque()
|
|
|
|
with ws._sub_lock:
|
|
current = list(ws._subscribed)
|
|
# 앞 soft 개는 유지, 나머지 REMOVE 후보
|
|
keep = current[:soft]
|
|
drop = current[soft:]
|
|
for c in keep:
|
|
active.append(c)
|
|
for c in drop:
|
|
overflow.append(c)
|
|
# subscribe_many 에서 클라이언트에 못 들어간 것도 overflow
|
|
for c in all_codes:
|
|
if c not in keep and c not in drop:
|
|
overflow.append(c)
|
|
|
|
if drop:
|
|
log.info("── Phase2: soft_cap=%d 초과 %d종 REMOVE ──", soft, len(drop))
|
|
for c in drop:
|
|
ws.unsubscribe(c)
|
|
time.sleep(2.0)
|
|
with ws._sub_lock:
|
|
n_sub = len(ws._subscribed)
|
|
log.info("REMOVE 후 subscribed=%d", n_sub)
|
|
|
|
# Phase 3: 시간 맞춰 로테이션
|
|
end_ts = time.time() + max(5, int(args.duration))
|
|
rot_every = int(args.rotate_every)
|
|
last_rot = time.time()
|
|
rot_round = 0
|
|
|
|
if rot_every <= 0:
|
|
log.info("로테이션 OFF — duration 동안 유지만")
|
|
else:
|
|
log.info(
|
|
"── Phase3: %ds 마다 %d종 교체 (overflow=%d) ──",
|
|
rot_every, rotate_n, len(overflow),
|
|
)
|
|
|
|
while time.time() < end_ts:
|
|
now = time.time()
|
|
if rot_every > 0 and (now - last_rot) >= rot_every and overflow:
|
|
rot_round += 1
|
|
last_rot = now
|
|
out_n = min(rotate_n, len(active), len(overflow))
|
|
if out_n <= 0:
|
|
log.info("로테이션 skip (active/overflow 부족)")
|
|
else:
|
|
leaving = [active.popleft() for _ in range(out_n)]
|
|
entering = [overflow.popleft() for _ in range(out_n)]
|
|
log.info(
|
|
"🔄 rotate#%d REMOVE %d → REG %d (active≈%d overflow=%d)",
|
|
rot_round, out_n, out_n, len(active), len(overflow),
|
|
)
|
|
for c in leaving:
|
|
ws.unsubscribe(c)
|
|
overflow.append(c)
|
|
time.sleep(0.5)
|
|
got = ws.subscribe_many(entering)
|
|
for c in got:
|
|
active.append(c)
|
|
time.sleep(1.5)
|
|
with ws._sub_lock:
|
|
n_sub = len(ws._subscribed)
|
|
log.info(
|
|
" 후 subscribed=%d reg_ok=%d reg_fail=%d ticks=%d",
|
|
n_sub, stats["reg_ok"], stats["reg_fail"], stats["real_ticks"],
|
|
)
|
|
|
|
# 1초마다 상태 한 줄
|
|
time.sleep(1.0)
|
|
if int(now) % 10 == 0:
|
|
with ws._sub_lock:
|
|
n_sub = len(ws._subscribed)
|
|
priced = 0
|
|
for c in list(active)[:soft]:
|
|
if ws.get_price(c, max_age_sec=30.0):
|
|
priced += 1
|
|
log.info(
|
|
"… hold subscribed=%d priced_sample≈%d/%d ticks=%d fail=%d",
|
|
n_sub, priced, min(len(active), soft),
|
|
stats["real_ticks"], stats["reg_fail"],
|
|
)
|
|
|
|
with ws._sub_lock:
|
|
n_sub = len(ws._subscribed)
|
|
log.info("=" * 70)
|
|
log.info("종료 요약")
|
|
log.info(" final_subscribed=%d", n_sub)
|
|
log.info(" login_ok=%d login_fail=%d", stats["login_ok"], stats["login_fail"])
|
|
log.info(" reg_ok=%d reg_fail=%d", stats["reg_ok"], stats["reg_fail"])
|
|
log.info(" real_ticks=%d", stats["real_ticks"])
|
|
if stats["reg_fail_msgs"]:
|
|
log.info(" fail_msgs 샘플:")
|
|
for m in stats["reg_fail_msgs"][:10]:
|
|
log.info(" - %s", m)
|
|
log.info(" 해석:")
|
|
log.info(" · Phase1 에서 reg_fail↑ / 세션 끊김 → 서버 한도≈100 근처")
|
|
log.info(" · REMOVE 후 REG 가 다시 ok → 로테이션 운용 가능")
|
|
log.info(" · 장외면 ticks=0 이어도 LOGIN/REG 로 한도 판정 가능")
|
|
log.info("로그 파일: %s", log_path)
|
|
log.info("=" * 70)
|
|
|
|
ws.stop()
|
|
time.sleep(1.0)
|
|
return 0 if stats["login_ok"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|