diff --git a/agents/stock/workspace/scripts/behive_web.py b/agents/stock/workspace/scripts/behive_web.py
index 5cb4c51..482bbf9 100644
--- a/agents/stock/workspace/scripts/behive_web.py
+++ b/agents/stock/workspace/scripts/behive_web.py
@@ -1845,9 +1845,9 @@ def _render_row(c: dict, source: str = 'watchlist') -> str:
if c.get('_show_day_change') and isinstance(day_pct, (int, float)):
d_sign = '+' if day_pct >= 0 else ''
d_cls = 'day-pct up' if day_pct > 0 else ('day-pct down' if day_pct < 0 else 'day-pct neutral')
- price_cell = f'{mark_html}{price:,}{d_sign}{day_pct:.2f}%'
+ price_cell = f'{mark_html}{price:,}{d_sign}{day_pct:.2f}%'
else:
- price_cell = f'{mark_html}{price:,}'
+ price_cell = f'{mark_html}{price:,}'
if mode == 'watching' and isinstance(ref_price, (int, float)) and ref_price > 0:
diff_cell = f'매수가 {ref_price:,.0f}'
else:
@@ -2157,9 +2157,12 @@ def _render_owner_kpi(owner: str, d: dict, owner_label_text: str, compact: bool
fee_html = f' · 수수료·세금 {realized_fees:,}원' if realized_fees else ''
realized_html = f'{rsign}{realized_pl:,}원{fee_html}'
kpis.append(('당일 실현손익', realized_html))
- rows = ''.join(
- f'
| {html.escape(k)} | {v} |
' for k, v in kpis
- )
+ _rtk = {'총 평가금액': 'value', '총 평가손익': 'profit', '순자산': 'net', '당일 평가손익': 'daypl'}
+ _row_list = []
+ for k, v in kpis:
+ attr = f' data-rt-kpi="{_rtk[k]}"' if k in _rtk else ''
+ _row_list.append(f'| {html.escape(k)} | {v} |
')
+ rows = ''.join(_row_list)
trade_btn_html = ''
if not compact:
owner_attr = html.escape(owner, quote=True)
@@ -2169,7 +2172,7 @@ def _render_owner_kpi(owner: str, d: dict, owner_label_text: str, compact: bool
f'data-trade-owner="{owner_attr}" data-trade-stock="{label_attr}">'
f'📋 전체 거래내역'
)
- return f'''
+ return f'''
{html.escape(owner_label_text)}
{trade_btn_html}
@@ -2569,7 +2572,7 @@ def _render_holding_row(r: dict, total_value: int, show_day_change: bool = False
)
pl_pairs: list[str] = [
- f'
평가손익{sign}{profit:,}원 ({sign}{profit_rate:.2f}%)',
+ f'
평가손익{sign}{profit:,}원 ({sign}{profit_rate:.2f}%)',
]
if day_change:
dcls = _profit_class(day_change)
@@ -2577,10 +2580,10 @@ def _render_holding_row(r: dict, total_value: int, show_day_change: bool = False
day_value = day_change * qty
dvsign = '+' if day_value >= 0 else ''
pl_pairs.append(
- f'
평가금액 변동{dvsign}{day_value:,}원'
+ f'
평가금액 변동{dvsign}{day_value:,}원'
)
pl_pairs.append(
- f'
당일 등락{dsign}{day_change:,}원 '
+ f'당일 등락{dsign}{day_change:,}원 '
f'({dsign}{day_change_pct:.2f}%)'
)
# 가격 출처 마크 — cur_price 가 어느 시장 응답 가격과 일치하는지로 판정.
@@ -2626,9 +2629,9 @@ def _render_holding_row(r: dict, total_value: int, show_day_change: bool = False
if show_day_change and isinstance(d_pct_val, (int, float)) and d_pct_val != 0:
d_sign = '+' if d_pct_val >= 0 else ''
d_cls = 'day-pct up' if d_pct_val > 0 else 'day-pct down'
- price_html = f'{mark_html}{price:,}{d_sign}{d_pct_val:.2f}%'
+ price_html = f'{mark_html}{price:,}{d_sign}{d_pct_val:.2f}%'
else:
- price_html = f'{mark_html}{price:,}'
+ price_html = f'{mark_html}{price:,}'
chart_block = ''
if code:
@@ -2644,7 +2647,7 @@ def _render_holding_row(r: dict, total_value: int, show_day_change: bool = False
right_pairs = [
f'보유수량{qty:,}주',
f'매입금액{buy_amount:,}원',
- f'평가금액{eval_value:,}원',
+ f'평가금액{eval_value:,}원',
]
dl_inner = _interleave_kv_pairs(left_pairs, right_pairs)
pl_pairs.extend(journal_lines)
@@ -2661,17 +2664,17 @@ def _render_holding_row(r: dict, total_value: int, show_day_change: bool = False
f' '
) if code else ''
tag_chip = _tag_chip_html(r.get('code') or '', r.get('stock') or '', interactive=True)
- return f'''
+ return f'''
{_buy_rank_badge(buy_amount)}{stock}{tag_chip}{star}{accounts}
-
보유 {qty:,}주{_pending_badges_html(r)}비중 {weight:.2f}%
+
보유 {qty:,}주{_pending_badges_html(r)}비중 {weight:.2f}%
{summary_journal_html}
{price_html}▾
평단 {avg:,}
-
{sign}{profit:,}{sign}{profit_rate:.2f}%
+
{sign}{profit:,}{sign}{profit_rate:.2f}%
{candle_summary}
@@ -3802,7 +3805,18 @@ def _render_owner_panel(owner: str, d: dict, balances: dict, owner_label_text: s
if not held and not phantoms:
parts.append('보유 종목 데이터가 없습니다.
')
- return '\n'.join(parts)
+ # rt-scope: 이 owner의 KPI 카드 + 보유행을 한 컨테이너로 묶음. 실시간 JS가 consolidated pane 행을
+ # 합산해 같은 scope 안의 data-rt-kpi(총평가금액·총평가손익·순자산·당일평가손익) td를 갱신.
+ # deposit은 순자산 계산용.
+ # 당일평가손익 = total_net − prev_net − net_cash_flow 에서 실시간 변동분은 평가액(total_value)뿐 →
+ # 당일평가손익 = 실시간평가액 + C, C = day_pl_total − total_value (세션 중 상수).
+ # prev_net은 % 계산용. day_pl_total None(전날 스냅 없음·휴장)이면 속성 미부착 → JS가 스킵.
+ deposit = d.get('deposit', 0) or 0
+ _daypl = d.get('day_pl_total')
+ _attrs = f'data-rt-owner="{html.escape(owner, quote=True)}" data-rt-deposit="{deposit}"'
+ if _daypl is not None:
+ _attrs += f' data-rt-daypl-c="{_daypl - d.get("total_value", 0)}" data-rt-prevnet="{d.get("prev_net", 0) or 0}"'
+ return f'' + '\n'.join(parts) + '
'
_CSS = '''
@@ -6194,8 +6208,9 @@ def render_html() -> str:
)
valid_tids_js = ','.join(f"'{tid}'" for tid in valid_tids)
- # 자동 갱신 동작 탭 — 자산정보·본인·가희(owner 부분 fetch)·관심종목(전체 fetch). 감시종목은 시세 변화 추적 대상 아니라 제외.
- auto_refresh_tids_js = ','.join(f"'{tid}'" for tid in valid_tids if tid != 'tab-wl')
+ # 자동 갱신 동작 탭 — 전체 탭(자산정보·본인·가희·관심·감시). 감시종목도 카운트다운 동작(2026-06-08 관리자 요청).
+ # 시세는 이미 실시간(rt-px)이라 풀패널 자동갱신은 미체결·시세외 데이터 재동기화 목적.
+ auto_refresh_tids_js = ','.join(f"'{tid}'" for tid in valid_tids)
# 탭 활성 동기화 — URL hash → html[data-tab]. CSS-only 전환이라 클릭은 즉시.
# 자산정보 탭으로 진입할 때는 전체 fetch 트리거 — 모든 owner를 한 화면에 모아 보는 요약 탭이라 stale 최소화.
@@ -8475,6 +8490,13 @@ function updateMarketPhaseDisplay(){
cls = 'closed';
canTrade = false;
}
+ // 실시간 VI 발동 종목이면 모달 상태줄에 표시 (window.__rtState 는 realtime_script 가 채움)
+ try {
+ var _rt = window.__rtState;
+ if(_rt && _rt.vi && state.code && _rt.vi[state.code] && _rt.vi[state.code].active){
+ label = '⚡VI 발동 · ' + label;
+ }
+ } catch(e){}
var el = $('[data-market-phase]');
if(el){
el.textContent = label + ' ' + (phase.time || '');
@@ -8905,6 +8927,131 @@ window.closeOrderModal = closeModal;
window.openPinModal = openPinModal;
})();'''
+ # 실시간 시세 갱신 — /api/realtime/quotes(허브 메모리) 1초 폴링으로 행 가격·VI 배지를 라이브 갱신.
+ # 허브 콜드/끊김이면 응답이 비어 아무것도 안 함(REST로 그려진 값 유지). visibility 가드.
+ realtime_script = r'''
+'''
+
return f'''
@@ -8964,6 +9111,7 @@ window.openPinModal = openPinModal;
{modal_script}
{order_modal_script}
{stock_name_tip_script}
+ {realtime_script}
'''
@@ -9144,6 +9292,34 @@ class Handler(BaseHTTPRequestHandler):
def do_GET(self): # noqa: N802
from urllib.parse import urlparse, parse_qs
parsed = urlparse(self.path)
+ if parsed.path == '/api/realtime/quotes':
+ # 실시간 허브 메모리의 최신 현재가/VI (키움 호출 0). 허브 콜드/끊김이면 빈 응답 → JS가 무시.
+ qs = parse_qs(parsed.query)
+ raw = (qs.get('codes') or [''])[0]
+ codes = [c for c in raw.split(',') if c.strip()]
+ quotes, vi, sub_count, connected = {}, {}, 0, False
+ hub = _RT_HUB
+ if hub is not None:
+ import kiwoom_client as kc
+ connected = hub.is_connected()
+ sub_count = len(hub._subs)
+ if not codes:
+ codes = sorted(hub._subs)
+ quotes = hub.get_quotes(codes)
+ for c in codes:
+ v = hub.get_vi(c)
+ if v:
+ vi[kc._clean_code(c)] = v
+ # 체결 신호 — fill_watcher가 체결 시 갱신. 브라우저가 값 변하면 패널 자동 새로고침.
+ fill_seq = ''
+ try:
+ _fp = WORKSPACE / 'state' / 'fill_signal'
+ if _fp.exists():
+ fill_seq = _fp.read_text().strip()
+ except Exception:
+ pass
+ self._send_json(200, {'connected': connected, 'sub_count': sub_count, 'quotes': quotes, 'vi': vi, 'fill_seq': fill_seq})
+ return
if parsed.path == '/api/trades':
qs = parse_qs(parsed.query)
code = (qs.get('code') or [''])[0].strip()
@@ -10180,7 +10356,90 @@ class Handler(BaseHTTPRequestHandler):
self.send_error(404)
+# ---------------- 실시간 시세 허브 (realtime_hub 연동) ----------------
+# 키움 WebSocket 실시간 현재가(0B)·VI(1h)를 메모리에 들고 화면에 라이브로 표시.
+# kill-switch: 환경변수 BEHIVE_REALTIME=0 이면 비활성(REST 폴백만). 기본 활성.
+# hub 시작·연결 실패가 페이지 서빙을 막지 않도록 전부 fail-safe.
+_RT_HUB = None
+_RT_MAX_CODES = 200 # 폭주 방지 상한. 키움은 59종목(=118등록) 수락 확인. 초과분은 우선순위 컷+로깅.
+
+
+def _rt_enabled() -> bool:
+ import os
+ return os.environ.get('BEHIVE_REALTIME', '1') not in ('0', 'false', 'no')
+
+
+def _rt_gather_codes() -> list:
+ """구독 대상: 보유(4계좌) > 관심 > 감시 우선순위. 상한 초과분은 잘라내고 로깅."""
+ import kiwoom_client as kc
+ seen: set = set()
+ codes: list = []
+
+ def _add(code):
+ if not code:
+ return
+ cc = kc._clean_code(code)
+ if cc and cc not in seen:
+ seen.add(cc)
+ codes.append(cc)
+
+ try:
+ for _label, positions in (kc.get_positions_all() or {}).items():
+ for p in positions:
+ _add(p.get('code'))
+ except Exception:
+ traceback.print_exc()
+ try:
+ for c in _load_interests():
+ _add(c.get('code'))
+ except Exception:
+ pass
+ try:
+ for c in _load_watchlist():
+ _add(c.get('code'))
+ except Exception:
+ pass
+ if len(codes) > _RT_MAX_CODES:
+ dropped = codes[_RT_MAX_CODES:]
+ sys.stderr.write(f'[realtime] 구독 상한 {_RT_MAX_CODES} 초과 → {len(dropped)}종목 제외: {dropped}\n')
+ codes = codes[:_RT_MAX_CODES]
+ return codes
+
+
+def _rt_resubscribe_loop():
+ while True:
+ try:
+ if _RT_HUB is not None:
+ codes = _rt_gather_codes()
+ if codes:
+ _RT_HUB.set_subscriptions(codes)
+ except Exception:
+ traceback.print_exc()
+ time.sleep(60)
+
+
+def _rt_start():
+ """실시간 허브 기동 (fail-safe). 실패해도 서버는 정상 서빙(REST 폴백)."""
+ global _RT_HUB
+ if not _rt_enabled():
+ print('[realtime] BEHIVE_REALTIME=0 → 실시간 허브 비활성 (REST 폴백)', flush=True)
+ return
+ try:
+ import realtime_hub as rh
+ _RT_HUB = rh.RealtimeHub()
+ codes = _rt_gather_codes()
+ _RT_HUB.set_subscriptions(codes)
+ _RT_HUB.start()
+ threading.Thread(target=_rt_resubscribe_loop, name='realtime-resub', daemon=True).start()
+ print(f'[realtime] 허브 기동 — {len(codes)}종목 구독', flush=True)
+ except Exception:
+ traceback.print_exc()
+ _RT_HUB = None
+ print('[realtime] 허브 기동 실패 → REST 폴백으로 계속', flush=True)
+
+
def cmd_serve() -> int:
+ _rt_start()
server = ThreadingHTTPServer((BIND_HOST, BIND_PORT), Handler)
print(f'serving on http://{BIND_HOST}:{BIND_PORT}', flush=True)
try:
diff --git a/agents/stock/workspace/scripts/orders/fill_watcher.py b/agents/stock/workspace/scripts/orders/fill_watcher.py
index 17a489d..c54c97d 100644
--- a/agents/stock/workspace/scripts/orders/fill_watcher.py
+++ b/agents/stock/workspace/scripts/orders/fill_watcher.py
@@ -61,6 +61,17 @@ _STATE_DIR = _WORKSPACE_ROOT / 'state'
QUEUE_FILE = _STATE_DIR / 'fill_pending.jsonl'
QUEUE_LOCK = _STATE_DIR / 'fill_pending.jsonl.lock'
PID_FILE = _STATE_DIR / 'fill_watcher.pid'
+# 체결 발생 신호 파일 — behive_web(별도 프로세스)이 /api/realtime/quotes 응답에 값을 실어주고,
+# 브라우저가 값 변화 감지 시 자산웹 패널을 자동 새로고침(수량·예수금·당일정산 반영). file 기반 IPC.
+FILL_SIGNAL_FILE = _STATE_DIR / 'fill_signal'
+
+
+def _bump_fill_signal() -> None:
+ """체결(전량/부분) 시 신호파일 갱신 → behive_web 패널 자동 새로고침 트리거. 실패 무시."""
+ try:
+ FILL_SIGNAL_FILE.write_text(str(time.time()), encoding='utf-8')
+ except Exception:
+ pass
# ---------- Tracked entry ----------
@@ -330,6 +341,7 @@ class _FillWatcher:
cntr_qty, cntr_uv, t.ord_no))
# 전량 체결 → 자산웹 거래내역 즉시 갱신 (별도 스레드, 추적 블로킹 X)
_spawn_journal_collect()
+ _bump_fill_signal() # 자산웹 패널 자동 새로고침(수량·예수금·당일정산)
else:
ledger.append('partial', {'card_id': t.card_id, 'ord_no': t.ord_no,
'account': t.account, 'symbol': t.symbol,
@@ -337,6 +349,7 @@ class _FillWatcher:
'price': cntr_uv, 'new_fill': new_fill})
self._send(card.format_partial(t.card_id, t.side, t.symbol_name,
cntr_qty, t.order_qty, cntr_uv, t.ord_no))
+ _bump_fill_signal() # 부분체결도 수량·예수금 변동 → 패널 새로고침
def _handle_timeout(self, t: Tracked) -> None:
with self._lock:
diff --git a/agents/stock/workspace/scripts/realtime_hub.py b/agents/stock/workspace/scripts/realtime_hub.py
new file mode 100644
index 0000000..8aacbf7
--- /dev/null
+++ b/agents/stock/workspace/scripts/realtime_hub.py
@@ -0,0 +1,318 @@
+#!/usr/bin/env python3
+"""키움 실시간 시세 허브 (WebSocket) — A단계 단독 모듈.
+
+조회 전용 REST(kiwoom_client)와 별개로, 키움 실시간 시세 WebSocket을 상시 청취해
+최신 현재가(0B 주식체결)·VI 발동/해제(1h)를 메모리에 들고 있다가 읽어주는 컴포넌트.
+
+behive_web 프로세스 안의 데몬 스레드로 띄울 예정이지만(B단계), 이 파일 자체는
+behive_web을 전혀 import/수정하지 않는다. 단독 실행으로 실접속·파싱을 먼저 검증한다.
+
+ python3 realtime_hub.py test 005930 000660 # 지정 종목 실시간 수신을 콘솔에 출력
+
+설계:
+ - 연결 1개(시세는 계좌 무관 → 토큰 아무 계좌나 재사용)
+ - 흐름: connect → LOGIN → REG(구독) → 수신 루프(PING 에코·REAL 파싱) → 끊기면 백오프 재연결+재구독
+ - 메모리 상태 저장소(lock): quotes{code:{...}}, vi{code:{...}}
+ - hub가 콜드/끊김이어도 호출측은 REST로 폴백 가능하도록, 읽기 메서드는 없으면 None 반환
+
+키움 WS 프로토콜:
+ - 엔드포인트: wss://api.kiwoom.com:10000/api/dostk/websocket (운영)
+ - LOGIN/PING 핸드셰이크는 공식 PDF에 표가 없어 키움 공식 샘플 규약을 따름:
+ LOGIN 송신: {"trnm":"LOGIN","token":""} (Bearer 접두어 없이 raw 토큰)
+ 서버가 {"trnm":"PING"} 주기 송신 → 같은 메시지 그대로 에코
+ - REG/REAL 스펙·필드코드는 PDF(국내주식>실시간시세) 기준:
+ 0B 주식체결: 10=현재가 11=전일대비 12=등락율 20=체결시간 16/17/18=시고저 9081=거래소
+ 1h VI발동/해제: 9068=발동구분 1221=발동가격 1224=해제시각 1225=적용구분 1489=등락율 9069=방향
+"""
+from __future__ import annotations
+
+import json
+import sys
+import threading
+import time
+from datetime import datetime, timezone, timedelta
+
+import websocket # websocket-client (동기식). 기존 threading 서버와 궁합.
+
+import kiwoom_client as kc
+
+KST = timezone(timedelta(hours=9))
+
+# 재연결 백오프 (초) — 점증, 상한 30s
+_BACKOFF_START = 2.0
+_BACKOFF_MAX = 30.0
+# 토큰 만료 전 선제 재연결 (초). 키움 토큰 ~30분 → 25분마다 새 연결로 갈아탐.
+_MAX_CONN_AGE = 25 * 60
+# recv 타임아웃 — 이 시간 내 아무것도 안 오면 루프 돌며 stop/age 체크
+_RECV_TIMEOUT = 5.0
+
+
+def _ws_url() -> str:
+ """REST base_url에서 WS 엔드포인트 도출. https→wss, :10000, /api/dostk/websocket."""
+ base = kc.base_url() # 예: https://api.kiwoom.com 또는 https://mockapi.kiwoom.com
+ host = base.split('://', 1)[-1].rstrip('/')
+ return f'wss://{host}:10000/api/dostk/websocket'
+
+
+def _abs_int(s) -> int:
+ """부호 포함 문자열 → 절대값 int (현재가·호가용)."""
+ return abs(kc._to_int(s))
+
+
+class RealtimeHub:
+ """키움 실시간 시세 허브. start()로 데몬 스레드 기동, get_quote/get_vi로 읽기."""
+
+ def __init__(self, account_label: str | None = None):
+ self._label = account_label or kc._default_account_label()
+ self._lock = threading.Lock()
+ self._quotes: dict[str, dict] = {} # code -> {price, change, pct, ts, exchange}
+ self._vi: dict[str, dict] = {} # code -> {active, kind, trigger_price, ...}
+ self._subs: set[str] = set() # 현재 구독 종목코드(정규화)
+ self._ws: websocket.WebSocket | None = None
+ self._ws_lock = threading.Lock() # send 직렬화
+ self._stop = threading.Event()
+ self._thread: threading.Thread | None = None
+ self._connected = threading.Event()
+ self.last_error: str | None = None
+ self.connected_at: float = 0.0
+
+ # ---- 공개 읽기 API (없으면 None → 호출측 REST 폴백) ----
+ def get_quote(self, code: str) -> dict | None:
+ with self._lock:
+ q = self._quotes.get(kc._clean_code(code))
+ return dict(q) if q else None
+
+ def get_quotes(self, codes: list[str]) -> dict[str, dict]:
+ with self._lock:
+ out = {}
+ for c in codes:
+ cc = kc._clean_code(c)
+ if cc in self._quotes:
+ out[cc] = dict(self._quotes[cc])
+ return out
+
+ def get_vi(self, code: str) -> dict | None:
+ with self._lock:
+ v = self._vi.get(kc._clean_code(code))
+ return dict(v) if v else None
+
+ def active_vis(self) -> dict[str, dict]:
+ with self._lock:
+ return {c: dict(v) for c, v in self._vi.items() if v.get('active')}
+
+ def is_connected(self) -> bool:
+ return self._connected.is_set()
+
+ # ---- 구독 관리 ----
+ def set_subscriptions(self, codes: list[str]) -> None:
+ """구독 종목 목록을 통째로 교체. 연결돼 있으면 즉시 REG/REMOVE 반영."""
+ new = {kc._clean_code(c) for c in codes if c}
+ with self._lock:
+ add = new - self._subs
+ rem = self._subs - new
+ self._subs = new
+ if self._connected.is_set():
+ if add:
+ self._send_reg(sorted(add))
+ if rem:
+ self._send_remove(sorted(rem))
+
+ # ---- 생명주기 ----
+ def start(self) -> None:
+ if self._thread and self._thread.is_alive():
+ return
+ self._stop.clear()
+ self._thread = threading.Thread(target=self._run, name='realtime-hub', daemon=True)
+ self._thread.start()
+
+ def stop(self) -> None:
+ self._stop.set()
+ with self._ws_lock:
+ if self._ws:
+ try:
+ self._ws.close()
+ except Exception:
+ pass
+
+ # ---- 내부: 연결 루프 ----
+ def _run(self) -> None:
+ backoff = _BACKOFF_START
+ while not self._stop.is_set():
+ try:
+ self._connect_once()
+ backoff = _BACKOFF_START # 정상 종료(선제 재연결 등) → 백오프 리셋
+ except Exception as e:
+ self.last_error = f'{type(e).__name__}: {e}'
+ self._connected.clear()
+ sys.stderr.write(f'[realtime-hub] 연결 끊김/실패: {self.last_error}\n')
+ if self._stop.is_set():
+ break
+ time.sleep(backoff)
+ backoff = min(backoff * 2, _BACKOFF_MAX)
+
+ def _connect_once(self) -> None:
+ url = _ws_url()
+ token = kc.issue_token(self._label) # 매 연결마다 fresh (캐시·만료는 issue_token이 관리)
+ ws = websocket.create_connection(url, timeout=10)
+ ws.settimeout(_RECV_TIMEOUT)
+ with self._ws_lock:
+ self._ws = ws
+ try:
+ # 1) LOGIN
+ ws.send(json.dumps({'trnm': 'LOGIN', 'token': token}))
+ login = json.loads(ws.recv())
+ if login.get('trnm') == 'LOGIN' and login.get('return_code') not in (0, '0', None):
+ raise RuntimeError(f'LOGIN 실패: {login.get("return_msg")} ({login})')
+ self._connected.set()
+ self.connected_at = time.time()
+ self.last_error = None
+ sys.stderr.write(f'[realtime-hub] LOGIN 성공 @ {url}\n')
+
+ # 2) 현재 구독 종목 REG
+ with self._lock:
+ codes = sorted(self._subs)
+ if codes:
+ self._send_reg(codes)
+
+ # 3) 수신 루프
+ while not self._stop.is_set():
+ if time.time() - self.connected_at > _MAX_CONN_AGE:
+ sys.stderr.write('[realtime-hub] 토큰 만료 전 선제 재연결\n')
+ return # 정상 종료 → _run이 즉시 재연결
+ try:
+ raw = ws.recv()
+ except websocket.WebSocketTimeoutException:
+ continue
+ if not raw:
+ continue
+ self._handle(raw)
+ finally:
+ self._connected.clear()
+ with self._ws_lock:
+ self._ws = None
+ try:
+ ws.close()
+ except Exception:
+ pass
+
+ # ---- 송신 ----
+ def _send(self, payload: dict) -> None:
+ with self._ws_lock:
+ if self._ws is None:
+ return
+ self._ws.send(json.dumps(payload))
+
+ def _send_reg(self, codes: list[str]) -> None:
+ # 0B(체결) + 1h(VI) 동시 등록. refresh=1 → 기존 구독 유지하며 추가.
+ self._send({
+ 'trnm': 'REG', 'grp_no': '1', 'refresh': '1',
+ 'data': [{'item': codes, 'type': ['0B', '1h']}],
+ })
+
+ def _send_remove(self, codes: list[str]) -> None:
+ self._send({
+ 'trnm': 'REMOVE', 'grp_no': '1',
+ 'data': [{'item': codes, 'type': ['0B', '1h']}],
+ })
+
+ # ---- 수신 처리 ----
+ def _handle(self, raw: str) -> None:
+ try:
+ msg = json.loads(raw)
+ except (ValueError, TypeError):
+ return
+ trnm = msg.get('trnm')
+ if trnm == 'PING':
+ # PING은 받은 그대로 에코 (키움 keepalive 규약)
+ with self._ws_lock:
+ if self._ws is not None:
+ self._ws.send(raw)
+ return
+ if trnm == 'REAL':
+ for blk in msg.get('data') or []:
+ self._handle_real(blk)
+ return
+ if trnm in ('REG', 'REMOVE'):
+ rc = msg.get('return_code')
+ if rc not in (0, '0', None):
+ sys.stderr.write(f'[realtime-hub] {trnm} 거부: rc={rc} msg={msg.get("return_msg")}\n')
+ else:
+ sys.stderr.write(f'[realtime-hub] {trnm} ok (rc={rc})\n')
+ return
+ # LOGIN 응답 등 기타는 무시
+
+ def _handle_real(self, blk: dict) -> None:
+ typ = blk.get('type')
+ code = kc._clean_code(blk.get('item') or '')
+ vals = blk.get('values') or {}
+ if not code:
+ return
+ if typ == '0B': # 주식체결 → 현재가
+ with self._lock:
+ self._quotes[code] = {
+ 'price': _abs_int(vals.get('10')),
+ 'change': kc._to_int(vals.get('11')),
+ 'pct': kc._to_float(vals.get('12')),
+ 'exchange': (vals.get('9081') or '').strip(),
+ 'ts': datetime.now(KST).isoformat(),
+ }
+ elif typ == '1h': # VI 발동/해제
+ # active 판정: 9068 VI발동구분. 해제시각(1224) 유무와 함께 실증 확인 예정(A단계).
+ kind_raw = (vals.get('9068') or '').strip()
+ with self._lock:
+ self._vi[code] = {
+ 'active': kind_raw not in ('', '0', '2'), # ⚠️ 매핑 실증 확인 필요
+ 'vi_gubun': kind_raw,
+ 'apply_kind': (vals.get('1225') or '').strip(), # 정적/동적/동적+정적
+ 'trigger_price': _abs_int(vals.get('1221')),
+ 'trigger_pct': kc._to_float(vals.get('1489')),
+ 'release_time': (vals.get('1224') or '').strip(),
+ 'direction': (vals.get('9069') or '').strip(),
+ 'ts': datetime.now(KST).isoformat(),
+ }
+
+
+# ---------------- CLI (A단계 단독 검증) ----------------
+def _cmd_test(codes: list[str]) -> int:
+ if not codes:
+ codes = ['005930', '000660'] # 삼성전자, SK하이닉스
+ print(f'[test] WS URL = {_ws_url()}')
+ print(f'[test] 구독 종목 = {codes} (Ctrl-C 종료)')
+ hub = RealtimeHub()
+ hub.set_subscriptions(codes)
+ hub.start()
+ last_print: dict[str, str] = {}
+ try:
+ while True:
+ time.sleep(1)
+ if not hub.is_connected():
+ continue
+ for c in codes:
+ cc = kc._clean_code(c)
+ q = hub.get_quote(cc)
+ if q:
+ line = f"{q['price']:,} ({q['pct']:+.2f}%) {q['exchange']}"
+ if last_print.get(cc) != line:
+ print(f' [{cc}] {line} @ {q["ts"][11:19]}')
+ last_print[cc] = line
+ v = hub.get_vi(cc)
+ if v and v.get('active'):
+ vk = f"VI:{cc} {v['apply_kind']} 발동가 {v['trigger_price']:,} ({v['trigger_pct']:+.2f}%)"
+ if last_print.get(f'vi_{cc}') != vk:
+ print(f' ⚡ {vk}')
+ last_print[f'vi_{cc}'] = vk
+ except KeyboardInterrupt:
+ print('\n[test] 종료')
+ hub.stop()
+ return 0
+
+
+def main(argv: list[str]) -> int:
+ if not argv or argv[0] != 'test':
+ print('usage: python3 realtime_hub.py test [code ...]', file=sys.stderr)
+ return 2
+ return _cmd_test(argv[1:])
+
+
+if __name__ == '__main__':
+ sys.exit(main(sys.argv[1:]))