""" kis_trader/strategies/updown_feed.py — UPDOWN(박스권) 시세 피드 어댑터 (ws | rest 스위치) ================================================================================ [역할] 박스권 watchlist 종목의 **현재가/당일 OHLC 스냅샷**을 전략 엔진에 공급한다. 공급원은 ``UPDOWN_FEED_MODE`` (env) 로 전환 — 전략 엔진 코드는 한 벌로 양쪽 다 동작. - ``rest`` (기본): KIS 멀티시세 ``intstock_multprice`` (1콜 ≤30종목) 를 백그라운드 스레드가 ``UPDOWN_REST_POLL_SEC`` (기본 5초) 마다 폴링해 캐시. watchlist 를 30 이하로 유지하면 **REST 1콜**로 전체 커버 → WS 슬롯 0. - ``ws`` : ``ws_manager`` (키움 WS) 에 watchlist 를 구독해 실시간 캐시 사용. 키움 100 한도(연결 총합) 가 모자라면(이미 모멘텀·돌파가 점유) **rest 폴백**. [중요 — 봉(candle) 은 여기서 안 준다] - 박스 판별·직전봉 음봉 판정용 **15분 확정봉**은 ``ws_manager.get_candles`` 가 담당 (WS 집계 또는 REST 폴백). 이 어댑터는 '현재가/하단근접/청산감시' 용 **현재가 스냅샷** 전용. - ``get_price(code)`` 는 KIS ``inquire_price`` 호환 dict (``stck_prpr`` 등) 로 정규화해 반환. 모든 수치는 하드코딩 금지 — ``get_env_*`` 로 DB/Env 로드. """ from __future__ import annotations import logging import threading import time from typing import Any, Callable, Dict, List, Optional from ..utils.env import get_env_from_db, get_env_int logger = logging.getLogger("kis_trader.updown_feed") # KIS 멀티시세 1콜 종목 한도 (intstock_multprice 스펙) _MULTI_PRICE_BATCH = 30 def feed_mode() -> str: """``UPDOWN_FEED_MODE`` — 'ws' 또는 'rest' (기본 rest). 그 외 값은 rest 로 폴백.""" v = (get_env_from_db("UPDOWN_FEED_MODE", "rest") or "rest").strip().lower() return "ws" if v == "ws" else "rest" def rest_poll_sec() -> float: """rest 모드 현재가 폴링 주기(초) — ``UPDOWN_REST_POLL_SEC`` (기본 5).""" return max(1, get_env_int("UPDOWN_REST_POLL_SEC", 5)) def _norm_multi_row(row: Dict[str, Any]) -> Dict[str, Any]: """intstock_multprice output → KIS inquire_price 호환 dict 로 정규화. 엔진/전략이 ``ws.get_price`` 에서 기대하는 키(``stck_prpr`` 등) 와 동일 포맷. 가격은 부호 제거(절대값) — 키움 WS 정규화와 동일 규약. """ def _absstr(v: Any) -> str: try: return str(int(abs(float(str(v).replace(",", ""))))) except (ValueError, TypeError): return "0" return { "stck_prpr": _absstr(row.get("inter2_prpr", "0")), "stck_oprc": _absstr(row.get("inter2_oprc", "0")), "stck_hgpr": _absstr(row.get("inter2_hgpr", "0")), "stck_lwpr": _absstr(row.get("inter2_lwpr", "0")), "stck_prdy_clpr": _absstr(row.get("inter2_prdy_clpr", "0")), "prdy_ctrt": str(row.get("prdy_ctrt", "0")), "acml_vol": _absstr(row.get("acml_vol", "0")), } class UpdownFeed: """UPDOWN 박스권 시세 피드 어댑터 (ws|rest 스위치 + rest 백그라운드 폴러).""" def __init__( self, *, ws_manager: Any, market_client: Any, codes_provider: Callable[[], List[str]], ): """ ws_manager: WSManager (get_price/get_candles/구독 위임). market_client: 실키 KISClient (inquire_multi_price 시세 호출 — .cursorrules 정책). codes_provider: 현재 active watchlist 코드 리스트를 반환하는 콜백 (예: lambda: [r['code'] for r in list_active_watchlist(db)]). """ self.ws = ws_manager self.market_client = market_client self._codes_provider = codes_provider # rest 캐시 — {code: {"data": {...}, "ts": epoch}} self._cache: Dict[str, Dict[str, Any]] = {} self._cache_lock = threading.Lock() # 백그라운드 폴러 self._poll_thread: Optional[threading.Thread] = None self._running = False self._mode = feed_mode() # ------------------------------------------------------------------ # 라이프사이클 # ------------------------------------------------------------------ def start(self) -> None: """rest 모드면 백그라운드 폴러 기동. ws 모드면 아무것도 안 함(ws_manager 가 담당).""" self._mode = feed_mode() if self._mode != "rest": logger.info("ℹ️ UpdownFeed: ws 모드 → 폴러 미기동(ws_manager 위임)") return if self._running: return self._running = True self._poll_thread = threading.Thread( target=self._poll_loop, daemon=True, name="UpdownFeedPoll", ) self._poll_thread.start() logger.info("✅ UpdownFeed: rest 폴러 시작 (멀티시세 %d배치, %.0f초 주기)", _MULTI_PRICE_BATCH, rest_poll_sec()) def stop(self) -> None: self._running = False # ------------------------------------------------------------------ # rest 폴러 # ------------------------------------------------------------------ def _poll_loop(self) -> None: while self._running: try: self._poll_once() except Exception as e: logger.debug("UpdownFeed 폴 루프 예외(무시): %s", e) # 주기 슬립 (env 핫리로드 반영 위해 매 루프 재조회) time.sleep(rest_poll_sec()) def _poll_once(self) -> None: codes = [str(c).strip() for c in (self._codes_provider() or []) if c] if not codes: return now = time.time() # 30 초과 시 배치 분할 (watchlist 30 이하면 1콜로 끝) for i in range(0, len(codes), _MULTI_PRICE_BATCH): batch = codes[i:i + _MULTI_PRICE_BATCH] try: rows = self.market_client.inquire_multi_price(batch) except Exception as e: logger.debug("inquire_multi_price 실패(배치 %d): %s", i, e) continue if not rows: continue with self._cache_lock: for code, raw in rows.items(): self._cache[code] = {"data": _norm_multi_row(raw), "ts": now} # ------------------------------------------------------------------ # 조회 인터페이스 (전략 엔진이 사용) # ------------------------------------------------------------------ def get_price(self, code: str, max_age_sec: Optional[float] = None) -> Optional[Dict[str, Any]]: """현재가 dict (KIS inquire_price 호환). 모드에 따라 ws 캐시 / rest 캐시 사용. max_age_sec is None 이면 나이 무시(마지막 RAM). 체결·폴러 공백 ≠ 가격 삭제. """ code = str(code).strip() if self._mode == "ws": try: return self.ws.get_price(code, max_age_sec=max_age_sec) except Exception: return None # rest 모드 — 폴러 캐시 with self._cache_lock: ent = self._cache.get(code) if not ent: return None if max_age_sec is not None and (time.time() - ent["ts"]) > max_age_sec: return None return ent["data"] def get_price_last(self, code: str) -> Optional[Dict[str, Any]]: return self.get_price(code, max_age_sec=None) def get_candles(self, code: str, tf: int, n: int = 50) -> list: """15분 등 확정봉 — 모드 무관 ws_manager 에 위임(WS 집계 또는 REST 폴백).""" try: return self.ws.get_candles(code, tf, n=n) except Exception: return [] # ------------------------------------------------------------------ # ws 모드 구독 동기화 (ws 모드일 때만 의미) # ------------------------------------------------------------------ def sync_ws_subscriptions(self, holdings: Optional[List[str]] = None) -> None: """ws 모드: active watchlist 를 키움 WS 후보로 구독 동기화. rest 모드면 no-op. 키움 슬롯이 모자라면 ws_manager 가 한도 초과분을 스킵하므로, 그 경우 rest 폴백을 권장(운영설정에서 UPDOWN_FEED_MODE=rest). """ if self._mode != "ws": return codes = [str(c).strip() for c in (self._codes_provider() or []) if c] hold = [str(h).strip() for h in (holdings or []) if h] try: self.ws.sync_targets_split("UPDOWN", codes, hold) except Exception as e: logger.debug("UPDOWN ws 구독 동기화 실패(무시): %s", e)