KIS Open API 조회전용 연동 + 직접매매 절대금지 안전게이트

매수/매도 주문 및 계좌 잔고조회를 API로 직접 실행하지 않는다는 원칙을 코드 레벨에서 강제하는 안전게이트(governance/rules/06, 07)와 함께, 시세/호가/공매도거래비중 등 조회전용 KIS Open API 연동 및 SQLite 수집 파이프라인을 추가한다. - kis_api_client_v1: 모든 요청이 _assert_read_only를 통과해야 하며 /trading/ 경로·주문 TR_ID는 RuntimeError로 즉시 차단 - kis_data_collection_v1: KIS 우선 + Naver 폴백, 네트워크 실패는 개별 ticker 단위로 흡수(배치 전체 중단 없음) - data_collection_store_v1 / storage_backend_v1: SQLite 캐노니컬 저장소, PostgreSQL 전환 대비 백엔드 추상화 - Gitea 영업일 스케줄(2시간 간격) + CI 강제 게이트 (validate_no_direct_api_trading_v1, validate_kis_api_credentials_v1)
2026-06-21 20:04:44 +09:00
parent 34f6eebba6
commit 4cb206a269
20 changed files with 2034 additions and 0 deletions
@@ -0,0 +1,106 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+try:
+    from src.quant_engine.kis_api_client_v1 import (
+        KisCredentials,
+        MOCK_DOMAIN,
+        REAL_DOMAIN,
+        _read_env_var,
+        get_current_price,
+    )
+except Exception as exc:  # pragma: no cover - import failure is a hard validation error
+    KisCredentials = None  # type: ignore[assignment]
+    MOCK_DOMAIN = ""
+    REAL_DOMAIN = ""
+    _read_env_var = None  # type: ignore[assignment]
+    get_current_price = None  # type: ignore[assignment]
+    _IMPORT_ERROR = str(exc)
+else:
+    _IMPORT_ERROR = ""
+
+
+def _payload(gate: str, **extra: Any) -> dict[str, Any]:
+    return {
+        "formula_id": "KIS_API_CREDENTIALS_VALIDATION_V1",
+        "gate": gate,
+        **extra,
+    }
+
+
+def _expected_env_names(account: str) -> tuple[str, str]:
+    if account == "real":
+        return ("KIS_APP_Key", "KIS_APP_Secret")
+    if account == "mock":
+        return ("KIS_APP_Key_TEST", "KIS_APP_Secret_TEST")
+    raise ValueError("account must be 'mock' or 'real'")
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="Validate KIS API credentials using the read-only quotations API.")
+    ap.add_argument("--account", choices=["mock", "real"], default="mock")
+    ap.add_argument("--ticker", default="005930")
+    ap.add_argument("--output", type=Path, default=ROOT / "Temp" / "kis_api_credentials_validation_v1.json")
+    args = ap.parse_args()
+
+    if KisCredentials is None or get_current_price is None:
+        result = _payload("FAIL", error=f"import_error: {_IMPORT_ERROR}")
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+        return 1
+
+    errors: list[str] = []
+    evidence: dict[str, Any] = {
+        "account": args.account,
+        "ticker": args.ticker,
+    }
+
+    try:
+        key_name, secret_name = _expected_env_names(args.account)
+        creds = KisCredentials.load(args.account)
+        evidence["domain"] = creds.domain
+        evidence["expected_env"] = {"app_key": key_name, "app_secret": secret_name}
+        expected_key = _read_env_var(key_name) if _read_env_var is not None else None
+        expected_secret = _read_env_var(secret_name) if _read_env_var is not None else None
+        other_key = _read_env_var("KIS_APP_Key_TEST" if args.account == "real" else "KIS_APP_Key") if _read_env_var is not None else None
+        other_secret = _read_env_var("KIS_APP_Secret_TEST" if args.account == "real" else "KIS_APP_Secret") if _read_env_var is not None else None
+        actual_key = getattr(creds, "app_key", None)
+        actual_secret = getattr(creds, "app_secret", None)
+        evidence["env_match"] = {
+            "app_key": bool(expected_key and actual_key == expected_key),
+            "app_secret": bool(expected_secret and actual_secret == expected_secret),
+            "other_key_present": bool(other_key),
+            "other_secret_present": bool(other_secret),
+        }
+        if creds.domain != (REAL_DOMAIN if args.account == "real" else MOCK_DOMAIN):
+            errors.append("domain_mismatch")
+        if not evidence["env_match"]["app_key"] or not evidence["env_match"]["app_secret"]:
+            errors.append("selected_env_mismatch")
+        response = get_current_price(creds, args.ticker)
+        evidence["response_keys"] = sorted(response.keys())
+        if not isinstance(response, dict) or not response:
+            errors.append("empty_response")
+    except Exception as exc:  # noqa: BLE001
+        errors.append(str(exc))
+
+    gate = "PASS" if not errors else "FAIL"
+    result = _payload(gate, evidence=evidence, errors=errors)
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 0 if gate == "PASS" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())