From 2cb95fc694a2f8cf787de4fbf9e7d058474f86d4 Mon Sep 17 00:00:00 2001 From: Hwang Date: Tue, 5 May 2026 21:39:34 +0900 Subject: [PATCH] =?UTF-8?q?feat(ws-validation):=20=ED=82=A4=EC=9B=80=20WS?= =?UTF-8?q?=20=EC=8B=9C=EC=84=B8=EB=8A=94=20KIS=5FMOCK=20=EB=AC=B4?= =?UTF-8?q?=EA=B4=80=20=ED=95=AD=EC=83=81=20=EC=8B=A4=ED=82=A4/=EC=8B=A4?= =?UTF-8?q?=EC=A0=84=20=EA=B0=95=EC=A0=9C=20(KIWOOM=5FWS=5FFORCE=5FREAL)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 배경 ---- - 모의 도메인(mockapi.kiwoom.com)은 장외 데이터 미제공 + 실시간 표본 빈약 → 시세 비교/마이그레이션 가치 없음 - 사용자 운영 정책: 키움은 실전 도메인에서 시간외 단일가까지 데이터 푸시 → 부하 회피용으로 매매 외 호출만 정지 토글 변경 ---- - database.py: ENV_CONFIG_KEYS 에 KIWOOM_WS_FORCE_REAL 추가 (기본 true) - kis_trader/main.py _start_ws_validator: 토글 true 면 _REAL 컬럼에서 직접 로드 + is_mock=False 강제, false 면 기존 _get_kiwoom_creds() 자동 분기 (디버그) - test_kiwoom_ws.py: load_kiwoom_creds(force_real=True) 기본, --mock-auto 플래그로 디버그 토글 영향 ---- - 매매(KIS) 100% 무영향 — 검증/테스트 채널만 실전 시세 도메인 강제 - KIS_MOCK=true 라도 시세 비교는 진짜 가격으로 진행 가능 Co-authored-by: Cursor --- database.py | 4 ++++ kis_trader/main.py | 39 +++++++++++++++++++++++++++++++++++---- test_kiwoom_ws.py | 40 +++++++++++++++++++++++++++++++++------- 3 files changed, 72 insertions(+), 11 deletions(-) diff --git a/database.py b/database.py index d89d317..8cc8034 100644 --- a/database.py +++ b/database.py @@ -428,6 +428,10 @@ ENV_CONFIG_KEYS = ( "WS_VALIDATION_INTERVAL_SEC", # 차이 경고 임계(%). |diff| 가 이 값 이상이면 WARN 로그. 기본 0.10. "WS_VALIDATION_DIFF_WARN_PCT", + # 키움 WS 시세는 KIS_MOCK 와 무관하게 항상 실키/실전 도메인 사용. + # 모의 도메인은 장외 데이터 제공 X / 실시간 표본도 빈약 → 시세 비교/마이그레이션 가치 없음. + # true (기본) = 실키 강제, false = KIS_MOCK 따라 자동 (디버그용). + "KIWOOM_WS_FORCE_REAL", ) diff --git a/kis_trader/main.py b/kis_trader/main.py index 322efbd..afc6c11 100644 --- a/kis_trader/main.py +++ b/kis_trader/main.py @@ -431,15 +431,46 @@ class TradingOrchestrator: ) return - # 키움 키 로드 (kis_ws._get_kiwoom_creds 재사용) + # 키움 키 로드. + # 정책: 시세 검증/마이그레이션은 항상 실키/실전 도메인을 사용해야 함. + # - 모의 도메인(mockapi.kiwoom.com) 은 장외 데이터 X, 실시간 표본도 빈약. + # - KIS_MOCK=true 라도 시세는 실전을 써야 진짜 가격 비교가 됨. + # KIWOOM_WS_FORCE_REAL=true (기본) 이면 DB env_config 의 _REAL 컬럼에서 직접 로드. + force_real_str = (get_env_from_db("KIWOOM_WS_FORCE_REAL", "true") or "true").strip().lower() + force_real = force_real_str in ("true", "1", "yes", "y") + + app_key = app_secret = "" + is_mock = False try: - from kis_ws import _get_kiwoom_creds # type: ignore - app_key, app_secret, is_mock = _get_kiwoom_creds(self.db) + if force_real: + row = self.db.conn.execute( + "SELECT * FROM env_config ORDER BY id DESC LIMIT 1" + ).fetchone() + if row: + r = dict(row) + app_key = (r.get("KIWOOM_APP_KEY_REAL") or "").strip() + app_secret = (r.get("KIWOOM_APP_SECRET_REAL") or "").strip() + if not app_key or not app_secret: + app_key = (r.get("KIWOOM_APP_KEY") or "").strip() + app_secret = (r.get("KIWOOM_APP_SECRET") or "").strip() + is_mock = False + logger.info("🔑 키움 WS 자격증명: 실키/실전 도메인 (KIWOOM_WS_FORCE_REAL=true)") + else: + from kis_ws import _get_kiwoom_creds # type: ignore + app_key, app_secret, is_mock = _get_kiwoom_creds(self.db) + logger.info( + "🔑 키움 WS 자격증명: KIS_MOCK 자동 (mock=%s) — 디버그 모드", + is_mock, + ) except Exception as e: logger.warning("키움 자격증명 로드 실패 → 검증 비활성: %s", e) return if not app_key or not app_secret: - logger.warning("키움 키 미설정 → WS 검증 비활성") + logger.warning( + "키움 %s 키 미설정 → WS 검증 비활성 (KIWOOM_APP_KEY%s 확인)", + "실키" if force_real else "키", + "_REAL" if force_real else "", + ) return # 키움 WS 인스턴스 diff --git a/test_kiwoom_ws.py b/test_kiwoom_ws.py index 6e2dbcc..ede3080 100644 --- a/test_kiwoom_ws.py +++ b/test_kiwoom_ws.py @@ -49,13 +49,32 @@ def setup_logging(verbose: bool) -> None: ) -def load_kiwoom_creds() -> tuple[str, str, bool]: - """DB env_config 에서 키움 키 로드 (KIS_MOCK 따라 MOCK/REAL 자동 선택).""" - from kis_ws import _get_kiwoom_creds # type: ignore +def load_kiwoom_creds(force_real: bool = True) -> tuple[str, str, bool]: + """DB env_config 에서 키움 키 로드. + + Args: + force_real: True (기본) 이면 KIS_MOCK 와 무관하게 항상 실키/실전 도메인. + 시세 검증·실거래 시세 비교에서는 모의 도메인이 의미 없으므로 권장. + False 면 ``_get_kiwoom_creds()`` 가 KIS_MOCK 따라 자동 선택 (디버그용). + """ 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_ws import _get_kiwoom_creds # type: ignore return _get_kiwoom_creds(db) finally: db.close() @@ -73,6 +92,10 @@ def main() -> int: help="구독 유지 시간(초). 기본 30", ) p.add_argument("-v", "--verbose", action="store_true", help="DEBUG 로그") + p.add_argument( + "--mock-auto", action="store_true", + help="KIS_MOCK 따라 자동 선택 (기본은 실키/실전 도메인 강제)", + ) args = p.parse_args() setup_logging(args.verbose) @@ -87,16 +110,19 @@ def main() -> int: print("🔍 키움 WS 단독 검증") print("=" * 70) - # 키 로드 + # 키 로드 (기본 실키 강제, --mock-auto 면 KIS_MOCK 따라 자동) + force_real = not args.mock_auto try: - app_key, app_secret, is_mock = load_kiwoom_creds() + app_key, app_secret, is_mock = load_kiwoom_creds(force_real=force_real) except Exception as e: print(f"❌ 키 로드 실패: {e}") return 1 if not app_key or not app_secret: - print("❌ 키움 키 미설정 (DB env_config 의 KIWOOM_APP_KEY_* 확인)") + col = "KIWOOM_APP_KEY_REAL" if force_real else "KIWOOM_APP_KEY_*" + print(f"❌ 키움 키 미설정 (DB env_config 의 {col} 확인)") return 1 - print(f" 키움 키: {app_key[:8]}…{app_key[-4:]} is_mock={is_mock}") + mode_label = "실전(force_real)" if force_real else f"auto (mock={is_mock})" + print(f" 키움 키: {app_key[:8]}…{app_key[-4:]} 도메인={mode_label}") print(f" 대상 종목: {codes}") print(f" 유지 시간: {args.duration}초") print()