kis_bot/kis_trader/scripts/show_account_snapshot.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
계좌 스냅샷 단독 조회 스크립트.

목적
----
- 봇 전체 로그 없이 계좌 핵심값(예수금총액/D+2/주문가능/평가금액)을 빠르게 확인.
- 모의/실전 어느 쪽 계좌를 보고 있는지 즉시 검증.

사용 예
--------
    python3 kis_trader/scripts/show_account_snapshot.py
    python3 kis_trader/scripts/show_account_snapshot.py --mode both
    python3 kis_trader/scripts/show_account_snapshot.py --mode real
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional

# 스크립트 직접 실행 시 프로젝트 루트 import 보정
HERE = Path(__file__).resolve()
ROOT = HERE.parents[2]  # /home/hoon/kis_bot
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from kis_trader.execution.kis_client import KISClient  # noqa: E402
from kis_trader.utils.env import get_env_bool, get_env_from_db  # noqa: E402


def _parse_amt(v: Any) -> float:
    try:
        return float(str(v or 0).replace(",", "").strip())
    except Exception:
        return 0.0


def _mask_account_no(raw: str) -> str:
    s = str(raw or "").strip()
    if not s:
        return "-"
    if len(s) <= 5:
        return s[0] + "***" + s[-1]
    return s[:3] + "***" + s[-2:]


def _pick_first_positive(d: Dict[str, Any], keys: List[str]) -> float:
    for k in keys:
        v = _parse_amt(d.get(k))
        if v > 0:
            return v
    return 0.0


def _extract_snapshot(balance: Dict[str, Any], client: KISClient) -> Dict[str, Any]:
    out2_raw = balance.get("output2") or []
    if isinstance(out2_raw, dict):
        out2 = out2_raw
    elif isinstance(out2_raw, list) and out2_raw:
        out2 = out2_raw[0]
    else:
        out2 = {}

    out1_raw = balance.get("output1") or []
    if isinstance(out1_raw, dict):
        out1 = [out1_raw]
    elif isinstance(out1_raw, list):
        out1 = out1_raw
    else:
        out1 = []

    dnca = _parse_amt(out2.get("dnca_tot_amt"))
    d2 = _parse_amt(out2.get("prvs_rcdl_excc_amt"))
    ord_psbl = _pick_first_positive(
        out2,
        ["ord_psbl_cash", "ord_psbl_amt", "ord_psbl", "buy_psbl_cash", "buy_psbl_amt"],
    )
    total_asset = _parse_amt(out2.get("tot_evlu_amt"))
    if total_asset <= 0:
        holding_eval = 0.0
        for it in out1:
            qty = _parse_amt(it.get("hldg_qty") or it.get("HLDG_QTY"))
            if qty <= 0:
                continue
            evlu = _parse_amt(it.get("evlu_amt") or it.get("EVLU_AMT"))
            if evlu > 0:
                holding_eval += evlu
                continue
            price = _parse_amt(it.get("prpr") or it.get("PRPR"))
            holding_eval += price * qty
        total_asset = (d2 if d2 > 0 else dnca) + holding_eval

    holdings = 0
    for it in out1:
        if _parse_amt(it.get("hldg_qty") or it.get("HLDG_QTY")) > 0:
            holdings += 1

    return {
        "mode": "MOCK" if client.mock else "REAL",
        "account_no": client.account_no or "",
        "account_code": client.account_code or "",
        "account_masked": f"{_mask_account_no(client.account_no)}-{client.account_code}",
        "dnca": dnca,
        "d2": d2,
        "ord_psbl": ord_psbl,
        "total_asset": total_asset,
        "holdings": holdings,
        "rt_cd": balance.get("rt_cd"),
        "msg_cd": str(balance.get("msg_cd", "") or "").strip(),
        "msg1": str(balance.get("msg1", "") or "").strip(),
    }


def _fetch_balance_raw(client: KISClient) -> Dict[str, Any]:
    """
    get_account_balance 래퍼 대신 raw 응답/HTTP 상태까지 수집.
    실패 원인 힌트를 주기 위해 msg_cd/msg1/status_code를 그대로 보존한다.
    """
    tr_id = "VTTC8434R" if client.mock else "TTTC8434R"
    params = {
        "CANO": client.account_no,
        "ACNT_PRDT_CD": client.account_code,
        "AFHR_FLPR_YN": "N",
        "OFL_YN": "N",
        "INQR_DVSN": "01",
        "UNPR_DVSN": "01",
        "FUND_STTL_ICLD_YN": "N",
        "FNCG_AMT_AUTO_RDPT_YN": "N",
        "PRCS_DVSN": "00",
        "CTX_AREA_FK100": "",
        "CTX_AREA_NK100": "",
    }
    try:
        resp = client._get(  # pylint: disable=protected-access
            "/uapi/domestic-stock/v1/trading/inquire-balance",
            tr_id,
            params,
        )
        status = int(getattr(resp, "status_code", 0) or 0)
        try:
            body = resp.json()
        except Exception:
            body = {}
        return {"ok": status == 200 and body.get("rt_cd") == "0", "status": status, "body": body}
    except Exception as e:
        return {"ok": False, "status": 0, "body": {}, "exception": str(e)}


def _diagnose_hint(row: Dict[str, Any]) -> List[str]:
    hints: List[str] = []
    status = int(row.get("http_status", 0) or 0)
    mode = str(row.get("mode", "")).upper()
    msg1 = str(row.get("msg1", "") or "")
    rt_cd = str(row.get("rt_cd", "") or "")
    account_code = str(row.get("account_code", "") or "")

    if status >= 500:
        hints.append("한투 서버/도메인 오류(HTTP 5xx). 계좌정보보다 API 상태 이슈 가능성이 큼")
    if mode == "MOCK" and status >= 500:
        hints.append("모의투자 잔고 API가 불안정할 수 있음. --mode real 결과와 비교 필요")
    if rt_cd != "0":
        hints.append("rt_cd!=0 이면 계좌번호/상품코드 조합 또는 권한/거래구분 불일치 가능")
    if "조회할 내용이 없습니다" in msg1:
        hints.append("계좌는 인식됐지만 조회 데이터가 비어있음: CANO/ACNT_PRDT_CD(보통 01) 재확인")
    if account_code not in ("01", "03"):
        hints.append("ACNT_PRDT_CD가 일반적 값(01/03)과 다름")
    if float(row.get("dnca", 0) or 0) <= 0 and float(row.get("total_asset", 0) or 0) <= 0:
        hints.append("응답상 현금/평가금 모두 0: 조회 대상 계좌가 다른 계좌일 가능성 높음")
    return hints


def _print_snapshot(s: Dict[str, Any]) -> None:
    print("=" * 72)
    print(f"[{s['mode']}] 계좌 {s['account_masked']}")
    print("-" * 72)
    print(f"HTTP 상태            : {int(s.get('http_status', 0))}")
    print(f"rt_cd/msg_cd        : {s.get('rt_cd', '-')}/{s.get('msg_cd', '-')}")
    print(f"예수금총액(dnca)   : {s['dnca']:>15,.0f} 원")
    print(f"D+2예수금(d2)      : {s['d2']:>15,.0f} 원")
    print(f"주문가능금액        : {s['ord_psbl']:>15,.0f} 원")
    print(f"총자산(tot_evlu)    : {s['total_asset']:>15,.0f} 원")
    print(f"보유종목수          : {int(s['holdings']):>15d} 개")
    if s.get("msg1"):
        print(f"응답메시지          : {s['msg1']}")
    hints = _diagnose_hint(s)
    if hints:
        print("힌트                :")
        for h in hints:
            print(f"  - {h}")


def _build_client(mode: str) -> KISClient:
    if mode == "mock":
        return KISClient(mock=True)
    if mode == "real":
        return KISClient(mock=False)
    # auto: env KIS_MOCK 사용
    return KISClient(mock=get_env_bool("KIS_MOCK", True))


def _resolve_modes(arg_mode: str) -> List[str]:
    m = (arg_mode or "auto").strip().lower()
    if m == "both":
        return ["mock", "real"]
    if m in ("mock", "real", "auto"):
        return [m]
    return ["auto"]


def main() -> None:
    p = argparse.ArgumentParser(description="계좌 핵심 스냅샷(예수금/주문가능/총자산) 조회")
    p.add_argument(
        "--mode",
        choices=["auto", "mock", "real", "both"],
        default="auto",
        help="조회 모드 (기본 auto=KIS_MOCK 따름)",
    )
    p.add_argument("--json", action="store_true", help="JSON 형식으로 출력")
    p.add_argument("--verbose", action="store_true", help="내부 HTTP/DB 로그 표시")
    args = p.parse_args()

    if not args.verbose:
        # 스냅샷 결과만 보이도록 내부 로거 소음 억제
        for name in (
            "TradeDB",
            "kis_trader.safe_request",
            "kis_token_manager",
            "kis_trader.kis_client",
        ):
            logging.getLogger(name).setLevel(logging.CRITICAL)

    rows: List[Dict[str, Any]] = []
    for mode in _resolve_modes(args.mode):
        client: Optional[KISClient] = None
        try:
            client = _build_client(mode)
            raw = _fetch_balance_raw(client)
            body = raw.get("body") or {}
            if not raw.get("ok"):
                rows.append(
                    {
                        "mode": "MOCK" if (client and client.mock) else "REAL",
                        "account_no": client.account_no if client else "",
                        "account_code": client.account_code if client else "",
                        "account_masked": (
                            f"{_mask_account_no(client.account_no)}-{client.account_code}"
                            if client
                            else "-"
                        ),
                        "http_status": int(raw.get("status", 0) or 0),
                        "rt_cd": str(body.get("rt_cd", "") or ""),
                        "msg_cd": str(body.get("msg_cd", "") or ""),
                        "msg1": str(body.get("msg1", "") or ""),
                        "error": (
                            str(raw.get("exception"))
                            if raw.get("exception")
                            else "잔고 조회 실패"
                        ),
                    }
                )
                continue
            snap = _extract_snapshot(body, client)
            snap["http_status"] = int(raw.get("status", 0) or 0)
            rows.append(snap)
        except Exception as e:
            rows.append(
                {
                    "mode": mode.upper(),
                    "account_no": client.account_no if client else "",
                    "account_code": client.account_code if client else "",
                    "account_masked": (
                        f"{_mask_account_no(client.account_no)}-{client.account_code}"
                        if client
                        else "-"
                    ),
                    "http_status": 0,
                    "error": str(e),
                }
            )

    if args.json:
        print(json.dumps(rows, ensure_ascii=False, indent=2))
        return

    print(f"KIS_MOCK(env) = {str(get_env_from_db('KIS_MOCK', 'true')).strip().lower()}")
    for r in rows:
        if r.get("error"):
            print("=" * 72)
            print(f"[{r.get('mode')}] 계좌 {r.get('account_masked')} 조회 실패")
            print(f"HTTP 상태            : {int(r.get('http_status', 0))}")
            print(f"rt_cd/msg_cd        : {r.get('rt_cd', '-')}/{r.get('msg_cd', '-')}")
            if r.get("msg1"):
                print(f"응답메시지          : {r.get('msg1')}")
            print(f"에러                : {r.get('error')}")
            hints = _diagnose_hint(r)
            if hints:
                print("힌트                :")
                for h in hints:
                    print(f"  - {h}")
        else:
            _print_snapshot(r)


if __name__ == "__main__":
    main()