[gpt와 함께하는 업무 자동화] 3. 키움증권 OpenAPI로 자동 트레이딩 시스템 만들기(삼성전자)

pythonKiwoomTrading.zip

0.01MB

 

1. kiwoomTradingBuy.py

# kiwoomTradingBuy.py
# -*- coding: utf-8 -*-

import json
import os
import time
from datetime import datetime
from typing import Any, Dict, Optional, List

import requests

# =========================================================
# ✅ 사용자 설정(여기만 수정)
# =========================================================
BASE_URL = "https://mockapi.kiwoom.com"
APP_KEY = "mAd3A.............0ZZuu8TxQ"
APP_SECRET = "hTo-..............E9ZgbeXozAU"

TARGET_STK_CD = "005930"
TARGET_STK_NM = "삼성전자"
EXCHANGE = "KRX"

# ✅ 기준가(구간 계산 기준) + 최초 매수 수량
BASE_PRICE_MANUAL = 166200
BUY_QTY = 100

# ✅ 요청 구간의 "경계값" (단위: %)
# 범위(0.01~0.05, 0.05~0.1, 0.1~0.8, 0.8~1, >=1)를 만들기 위한 경계값 목록
THRESHOLDS: List[float] = [0.01, 0.05, 0.1, 0.8, 1.0]

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATE_PATH = os.path.join(BASE_DIR, "trade_state.json")
LOG_PATH = os.path.join(BASE_DIR, "trade_log.txt")

TIMEOUT_SEC = 15

# ✅ 주문 API(kt10000) 연속 호출 방지 쿨다운(초)
ORDER_COOLDOWN_SEC = 8


# =========================================================
# 유틸
# =========================================================
def now_str() -> str:
    """현재 시간을 'YYYY-mm-dd HH:MM:SS' 문자열로 반환"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def now_epoch() -> float:
    """현재 시간을 epoch(float)로 반환 (쿨다운 계산용)"""
    return time.time()


def log_line(title: str, msg: str) -> None:
    """
    [기능] 로그를 파일(trade_log.txt)과 콘솔에 남김
    - msg에 개행(\n)이 있어도 그대로 출력/저장
    """
    line = f"{now_str()} | {title} | {msg}\n"
    with open(LOG_PATH, "a", encoding="utf-8") as f:
        f.write(line)
    print(line.strip())


def save_json_atomic(path: str, obj: Dict[str, Any]) -> None:
    """
    [기능] JSON을 임시파일(.tmp)로 저장 후 원자적 교체(os.replace)
    - 저장 중 크래시가 나도 파일이 깨질 확률을 낮춤
    """
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False, indent=2)
    os.replace(tmp, path)


def load_state(path: str) -> Dict[str, Any]:
    """
    [기능] trade_state.json 로드
    - 없거나 깨져 있으면 {} 반환
    """
    if not os.path.exists(path):
        return {}
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def pct_key(x: float) -> str:
    """
    [기능] bands/thresholds 키로 쓸 문자열을 안정적으로 생성
    - 1.0 -> "1"
    - 0.01 -> "0.01"
    """
    if abs(x - int(x)) < 1e-9:
        return str(int(x))
    return f"{x:.6f}".rstrip("0").rstrip(".")


def fmt_pct(x: float) -> str:
    """
    [기능] 퍼센트 숫자를 보기 좋게 표시
    - 1.0 -> "1.0%"
    - 0.01 -> "0.01%"
    """
    if abs(x - int(x)) < 1e-9:
        return f"{int(x)}.0%"
    s = f"{x:.6f}".rstrip("0").rstrip(".")
    return f"{s}%"


def calc_levels(base_price: int, thresholds: List[float]) -> Dict[str, Any]:
    """
    [기능] 기준가(base_price)와 thresholds(경계값 %)로 구간 가격(up/down)을 계산해 state에 저장할 구조 생성
    - bands["0.01"]["up"] / bands["0.01"]["down"] 형태
    """
    th = sorted(list(set([float(x) for x in thresholds])))
    levels: Dict[str, Any] = {
        "base_price": int(base_price),
        "thresholds": th,  # float 그대로 저장
        "bands": {},
    }

    for pct in th:
        up = int(round(base_price * (1 + pct / 100.0)))
        down = int(round(base_price * (1 - pct / 100.0)))
        levels["bands"][pct_key(pct)] = {"up": up, "down": down}

    return levels


def format_base_levels_for_log(base_price: int, thresholds: List[float], bands: Dict[str, Any]) -> str:
    """
    [기능] 기준가 및 bands를 여러 줄로 예쁘게 로그 출력하기 위한 문자열 생성
    """
    th = sorted(list(set([float(x) for x in thresholds])))

    lines: List[str] = []
    lines.append(f"기준가(base) : {base_price:,}원")
    lines.append("")
    lines.append("▲ 상승 경계값(참고용)")
    for t in th:
        key = pct_key(t)
        up = int(bands[key]["up"])
        lines.append(f"  · {fmt_pct(t)} → {up:,}원")

    lines.append("")
    lines.append("▼ 하락 경계값(참고용)")
    for t in th:
        key = pct_key(t)
        dn = int(bands[key]["down"])
        lines.append(f"  · {fmt_pct(t)} → {dn:,}원")

    lines.append("")
    lines.append("※ 실제 매매 판단은 BuySell에서 '퍼센트 범위(구간)'로 판정합니다.")
    return "\n".join(lines)


# =========================================================
# 예외 정의
# =========================================================
class KiwoomAPIError(RuntimeError):
    """[기능] 키움 API가 return_code != 0 응답을 준 경우를 예외로 래핑"""

    def __init__(self, data: Dict[str, Any]):
        self.data = data
        super().__init__(f"API 오류: {data}")

    @property
    def return_code(self) -> Optional[int]:
        """[기능] return_code를 int로 반환 (없으면 None)"""
        try:
            return int(self.data.get("return_code"))
        except Exception:
            return None

    @property
    def return_msg(self) -> str:
        """[기능] return_msg 문자열 반환"""
        return str(self.data.get("return_msg") or "")


# =========================================================
# Kiwoom REST Client
# =========================================================
class KiwoomClient:
    """[기능] 키움 REST API 호출(토큰 발급/주문) 담당"""

    def __init__(self, base_url: str, app_key: str, app_secret: str):
        self.base_url = base_url.rstrip("/")
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = requests.Session()
        self.token: Optional[str] = None

    def authenticate(self) -> str:
        """[기능] OAuth2 토큰 발급 후 self.token에 저장"""
        url = f"{self.base_url}/oauth2/token"
        body = {"grant_type": "client_credentials", "appkey": self.app_key, "secretkey": self.app_secret}
        headers = {"Content-Type": "application/json;charset=UTF-8"}
        resp = self.session.post(url, headers=headers, json=body, timeout=TIMEOUT_SEC)
        data = resp.json()
        if resp.status_code >= 400 or (isinstance(data, dict) and data.get("return_code") not in (None, 0)):
            raise RuntimeError(f"토큰 발급 실패: HTTP {resp.status_code} / {data}")
        self.token = "Bearer " + str(data["token"])
        return self.token

    def post(self, path: str, api_id: str, body: Dict[str, Any]) -> Dict[str, Any]:
        """[기능] 공통 POST 호출 + 에러 처리(return_code 포함)"""
        if not self.token:
            raise RuntimeError("토큰이 없습니다. authenticate() 먼저 호출하세요.")
        url = f"{self.base_url}{path}"
        headers = {"Content-Type": "application/json;charset=UTF-8", "authorization": self.token, "api-id": api_id}
        resp = self.session.post(url, headers=headers, json=body, timeout=TIMEOUT_SEC)

        try:
            data = resp.json()
        except Exception:
            raise RuntimeError(f"HTTP {resp.status_code}: 응답이 JSON이 아닙니다. text={resp.text[:300]}")

        if resp.status_code >= 400:
            raise RuntimeError(f"HTTP {resp.status_code}: {data}")

        if isinstance(data, dict) and data.get("return_code") not in (None, 0):
            raise KiwoomAPIError(data)

        return data

    def buy_market(self, stk_cd: str, qty: int) -> Dict[str, Any]:
        """[기능] 시장가 매수 주문 제출(kt10000)"""
        body = {
            "dmst_stex_tp": EXCHANGE,
            "stk_cd": stk_cd,
            "ord_qty": str(qty),
            "trde_tp": "3",  # 3=시장가
        }
        return self.post("/api/dostk/ordr", api_id="kt10000", body=body)


def main() -> None:
    """[기능] 최초 1회 시장가 매수 + trade_state.json 생성(구간 경계값/bands 포함)"""
    prev_state = load_state(STATE_PATH)
    last_api_call = float(prev_state.get("meta", {}).get("last_api_call_epoch") or 0.0)
    since = now_epoch() - last_api_call

    if last_api_call > 0 and since < ORDER_COOLDOWN_SEC:
        remain = ORDER_COOLDOWN_SEC - since
        log_line("RATE_LIMIT_GUARD", f"직전 주문 API 호출 후 {since:.1f}s 경과. {remain:.1f}s 더 기다린 후 재실행하세요.")
        return

    cli = KiwoomClient(BASE_URL, APP_KEY, APP_SECRET)
    cli.authenticate()

    base_price = int(BASE_PRICE_MANUAL)
    ord_no = ""
    note = ""

    # ✅ 이번 실행에서 주문 API를 호출하므로, 호출 직전에 기록(크래시 대비)
    meta = prev_state.get("meta", {})
    meta.update({"updated_at": now_str(), "last_api_call_epoch": now_epoch()})
    prev_state["meta"] = meta
    save_json_atomic(STATE_PATH, prev_state)

    try:
        log_line("TRY_MARKET", f"{TARGET_STK_NM}({TARGET_STK_CD}) 시장가 매수 시도 | qty={BUY_QTY} | 기준가(구간계산)={base_price:,}원")
        resp = cli.buy_market(TARGET_STK_CD, BUY_QTY)
        ord_no = str(resp.get("ord_no") or "").strip()
        note = "시장가로 주문(모의투자 제한 회피)"

    except KiwoomAPIError as e:
        msg = e.return_msg
        code = e.return_code

        if code == 5 or "1700" in msg:
            log_line("RATE_LIMIT_HIT", f"허용된 요청 개수 초과로 중단합니다(추가 호출 금지). 상세={e.data}")
            return

        log_line("ORDER_FAILED", f"시장가 주문 실패 | return_code={code} | msg={msg} | raw={e.data}")
        raise

    if not ord_no:
        log_line("NO_ORD_NO", "주문번호(ord_no)가 비어있습니다. 응답을 확인하세요.")
        return

    log_line(
        "BUY_SUBMIT",
        f"{TARGET_STK_NM}({TARGET_STK_CD}) 주문 제출 완료 | type=MARKET_ONLY | qty={BUY_QTY} | base_price={base_price:,} | ord_no={ord_no}",
    )

    levels = calc_levels(base_price, THRESHOLDS)

    state: Dict[str, Any] = {
        "meta": {
            "created_at": prev_state.get("meta", {}).get("created_at") or now_str(),
            "updated_at": now_str(),
            "last_api_call_epoch": meta.get("last_api_call_epoch"),
        },
        "last_order": {
            "ts": now_str(),
            "stk_cd": TARGET_STK_CD,
            "stk_nm": TARGET_STK_NM,
            "side": "BUY",
            "ord_no": ord_no,
            "qty_req": BUY_QTY,
            "price_req": base_price,   # 기준가(구간 계산 기준)
            "price_sent": None,        # 시장가는 지정가 없음
            "order_type": "MARKET_ONLY",
            "note": note,
        },
        "base": levels,
        "triggers_done": {"up": {}, "down": {}},  # BuySell에서 구간별 1회 실행 체크
        "virtual_position": {"stk_cd": TARGET_STK_CD, "stk_nm": TARGET_STK_NM, "qty": BUY_QTY},
    }

    save_json_atomic(STATE_PATH, state)

    pretty = format_base_levels_for_log(
        base_price=state["base"]["base_price"],
        thresholds=state["base"]["thresholds"],
        bands=state["base"]["bands"],
    )
    log_line("BASE_LEVELS", f"\n{pretty}")
    log_line("STATE_SAVED", f"trade_state.json 저장 완료: {STATE_PATH}")


if __name__ == "__main__":
    main()

2. kiwoomTradingBuySell.py

# kiwoomTradingBuySell.py
# -*- coding: utf-8 -*-
"""
[초보자용 - 자동매매 1액션 (퍼센트 "범위" 기준 버전)]

요청 규칙(범위):
[하락]
- 0.01% 이상 ~ 0.05% 미만 떨어짐  -> 20주 매수
- 0.05% 이상 ~ 0.1% 미만 떨어짐  -> 20주 매수
- 0.1% 이상 ~ 0.8% 미만 떨어짐   -> 20주 매수
- 0.8% 이상 ~ 1% 미만 떨어짐     -> 20주 매수
- 1% 이상 떨어짐                  -> 20주 매수

[상승]
- 0.01% 이상 ~ 0.05% 미만 오름  -> 20주 매도
- 0.05% 이상 ~ 0.1% 미만 오름   -> 20주 매도
- 0.1% 이상 ~ 0.8% 미만 오름    -> 20주 매도
- 0.8% 이상 ~ 1% 미만 오름      -> 20주 매도
- 1% 이상 오름                   -> 20주 매도

판정 방식(핵심):
- 기준가(base_price) 대비 '퍼센트 변화율'을 직접 계산해서 "범위(구간)"로 판정합니다.
- 상승(매도): 매수호가(buy_price) 기준 pct_up = (buy_price/base - 1)*100
- 하락(매수): 매도호가(sell_price) 기준 pct_down = (1 - sell_price/base)*100

주의:
- 키움 호가가 "-150900"처럼 음수로 내려오는 케이스가 있어 abs()로 가격을 양수 정규화합니다.
"""

import json
import os
from datetime import datetime
from typing import Any, Dict, Optional, Tuple, List

import requests

# =========================
# 환경 설정
# =========================
BASE_URL = "https://mockapi.kiwoom.com"
APP_KEY = "mAd3AP................Zuu8TxQ"
APP_SECRET = "hTo-Z.................beXozAU"

TARGET_STK_CD = "005930"
TARGET_STK_NM = "삼성전자"
EXCHANGE = "KRX"

TRADE_QTY = 20
TIMEOUT_SEC = 15

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
LOG_PATH = os.path.join(BASE_DIR, "trade_log.txt")
STATE_PATH = os.path.join(BASE_DIR, "trade_state.json")

# ✅ STATE에 thresholds가 없을 때 fallback(경계값)
DEFAULT_THRESHOLDS: List[float] = [0.01, 0.05, 0.1, 0.8, 1.0]


# =========================
# 유틸
# =========================
def now_str() -> str:
    """[기능] 현재 시간 문자열 반환"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def log_line(cat: str, msg: str) -> None:
    """[기능] 로그를 파일+콘솔에 기록(사람이 읽기 쉬운 한글 문장)"""
    line = f"{now_str()} | {cat} | {msg}\n"
    with open(LOG_PATH, "a", encoding="utf-8") as f:
        f.write(line)
    print(line.strip())


def load_state() -> Dict[str, Any]:
    """[기능] trade_state.json 로드(없으면 종료)"""
    if not os.path.exists(STATE_PATH):
        raise SystemExit(f"state 파일이 없습니다: {STATE_PATH} (먼저 kiwoomTradingBuy.py 실행)")
    with open(STATE_PATH, "r", encoding="utf-8") as f:
        return json.load(f)


def save_state(state: Dict[str, Any]) -> None:
    """[기능] trade_state.json 원자적 저장 + meta.updated_at 갱신"""
    state.setdefault("meta", {})
    state["meta"]["updated_at"] = now_str()

    tmp = STATE_PATH + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)
    os.replace(tmp, STATE_PATH)


def to_int(x) -> Optional[int]:
    """[기능] 숫자형 문자열/숫자를 int로 변환(콤마/float/음수 포함)"""
    try:
        if x is None:
            return None
        s = str(x).strip().replace(",", "")
        if s == "":
            return None
        return int(float(s))
    except Exception:
        return None


def to_price_int(x) -> Optional[int]:
    """[기능] 가격 값은 abs()로 양수 정규화(키움이 '-150900'처럼 주는 케이스 방어)"""
    v = to_int(x)
    return abs(v) if v is not None else None


def fmt_won(x: Optional[int]) -> str:
    """[기능] 원화 표시 포맷"""
    return f"{x:,}원" if isinstance(x, int) else "N/A"


def fmt_pct_value(x: float, digits: int = 4) -> str:
    """[기능] 퍼센트 수치를 보기 좋게 (예: 0.0375 -> '0.0375%')"""
    s = f"{x:.{digits}f}".rstrip("0").rstrip(".")
    return f"{s}%"


def parse_thresholds(base: Dict[str, Any]) -> List[float]:
    """[기능] base.thresholds를 float list로 정규화(없으면 DEFAULT_THRESHOLDS)"""
    raw = base.get("thresholds")
    if not raw:
        return DEFAULT_THRESHOLDS[:]
    out: List[float] = []
    for v in raw:
        try:
            out.append(float(str(v).strip()))
        except Exception:
            continue
    out = sorted(list(set(out)))
    return out if out else DEFAULT_THRESHOLDS[:]


def build_ranges_from_thresholds(thresholds: List[float]) -> List[Tuple[float, Optional[float], str]]:
    """
    [기능] 경계값(thresholds)으로 '범위 구간' 목록을 만든다.
    예) [0.01, 0.05, 0.1, 0.8, 1.0] ->
      (0.01, 0.05, '0.01~0.05')
      (0.05, 0.1,  '0.05~0.1')
      (0.1,  0.8,  '0.1~0.8')
      (0.8,  1.0,  '0.8~1')
      (1.0,  None, '>=1')
    """
    th = sorted(list(set([float(x) for x in thresholds])))
    ranges: List[Tuple[float, Optional[float], str]] = []
    if not th:
        th = DEFAULT_THRESHOLDS[:]

    for i in range(len(th)):
        lo = th[i]
        hi = th[i + 1] if i + 1 < len(th) else None
        if hi is None:
            key = f">={strip_trailing_zeros(lo)}"
        else:
            key = f"{strip_trailing_zeros(lo)}~{strip_trailing_zeros(hi)}"
        ranges.append((lo, hi, key))
    return ranges


def strip_trailing_zeros(x: float) -> str:
    """[기능] 1.0 -> '1', 0.10 -> '0.1', 0.01 -> '0.01'"""
    if abs(x - int(x)) < 1e-12:
        return str(int(x))
    return f"{x:.6f}".rstrip("0").rstrip(".")


def classify_pct_to_range_key(pct: float, ranges: List[Tuple[float, Optional[float], str]]) -> Optional[str]:
    """
    [기능] pct(퍼센트 변화율)를 ranges 중 어디에 속하는지 찾아 '구간키' 반환
    - lo 이상, hi 미만
    - 마지막 구간은 lo 이상(>=lo)
    """
    for lo, hi, key in ranges:
        if hi is None:
            if pct >= lo:
                return key
        else:
            if pct >= lo and pct < hi:
                return key
    return None


def calc_pct_up(base_price: int, price: int) -> float:
    """[기능] 상승 퍼센트 계산: (price/base - 1)*100"""
    if base_price <= 0:
        return 0.0
    return (price / base_price - 1.0) * 100.0


def calc_pct_down(base_price: int, price: int) -> float:
    """[기능] 하락 퍼센트 계산: (1 - price/base)*100"""
    if base_price <= 0:
        return 0.0
    return (1.0 - price / base_price) * 100.0


def fmt_range_label(direction: str, range_key: str) -> str:
    """[기능] 로그용 구간 라벨(상승/하락 + 범위키)"""
    return f"{'하락' if direction == 'down' else '상승'} {range_key}% 구간"


# =========================
# 예외(리밋코드 등 처리용)
# =========================
class KiwoomAPIError(RuntimeError):
    """[기능] return_code 기반 에러 처리용"""

    def __init__(self, data: Dict[str, Any]):
        self.data = data
        super().__init__(f"API 오류: {data}")

    @property
    def return_code(self) -> Optional[int]:
        try:
            return int(self.data.get("return_code"))
        except Exception:
            return None

    @property
    def return_msg(self) -> str:
        return str(self.data.get("return_msg") or "")


# =========================
# 키움 REST 클라이언트
# =========================
class KiwoomClient:
    """[기능] 토큰 발급 + 호가조회 + 시장가 매수/매도 주문"""

    def __init__(self, base_url: str, app_key: str, app_secret: str):
        self.base_url = base_url.rstrip("/")
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = requests.Session()
        self.token: Optional[str] = None

    def authenticate(self) -> str:
        """[기능] OAuth2 토큰 발급"""
        url = f"{self.base_url}/oauth2/token"
        body = {"grant_type": "client_credentials", "appkey": self.app_key, "secretkey": self.app_secret}
        headers = {"Content-Type": "application/json;charset=UTF-8"}
        resp = self.session.post(url, headers=headers, json=body, timeout=TIMEOUT_SEC)
        data = resp.json()
        if resp.status_code >= 400 or (isinstance(data, dict) and data.get("return_code") not in (None, 0)):
            raise RuntimeError(f"토큰 발급 실패: HTTP {resp.status_code} / {data}")
        self.token = "Bearer " + str(data["token"])
        return self.token

    def post(self, path: str, api_id: str, body: Dict[str, Any]) -> Dict[str, Any]:
        """[기능] 공통 POST 호출 + return_code 처리"""
        if not self.token:
            raise RuntimeError("토큰이 없습니다. authenticate() 먼저 호출하세요.")
        url = f"{self.base_url}{path}"
        headers = {"Content-Type": "application/json;charset=UTF-8", "authorization": self.token, "api-id": api_id}
        resp = self.session.post(url, headers=headers, json=body, timeout=TIMEOUT_SEC)

        try:
            data = resp.json()
        except Exception:
            raise RuntimeError(f"HTTP {resp.status_code}: 응답이 JSON이 아닙니다. text={resp.text[:300]}")

        if resp.status_code >= 400:
            raise RuntimeError(f"HTTP {resp.status_code}: {data}")

        if isinstance(data, dict) and data.get("return_code") not in (None, 0):
            raise KiwoomAPIError(data)

        return data

    def get_best_prices(self, stk_cd: str) -> Tuple[Optional[int], Optional[int]]:
        """
        [기능] 최우선 호가 조회 (ka10004)
        - 매수호가(최우선 매수): buy_fpr_bid
        - 매도호가(최우선 매도): sel_fpr_bid
        """
        body = {"stk_cd": stk_cd}
        data = self.post("/api/dostk/mrkcond", api_id="ka10004", body=body)

        buy_price = to_price_int(data.get("buy_fpr_bid"))
        sell_price = to_price_int(data.get("sel_fpr_bid"))
        return buy_price, sell_price

    def buy_market(self, stk_cd: str, qty: int) -> Dict[str, Any]:
        """[기능] 시장가 매수 주문(kt10000)"""
        body = {"dmst_stex_tp": EXCHANGE, "stk_cd": stk_cd, "ord_qty": str(qty), "trde_tp": "3"}
        return self.post("/api/dostk/ordr", api_id="kt10000", body=body)

    def sell_market(self, stk_cd: str, qty: int) -> Dict[str, Any]:
        """[기능] 시장가 매도 주문(kt10001)"""
        body = {"dmst_stex_tp": EXCHANGE, "stk_cd": stk_cd, "ord_qty": str(qty), "trde_tp": "3"}
        return self.post("/api/dostk/ordr", api_id="kt10001", body=body)


# =========================
# 구간(범위) 판정 로직
# =========================
def decide_range_action(
    base_price: int,
    buy_price: Optional[int],
    sell_price: Optional[int],
    thresholds: List[float],
) -> Tuple[Optional[str], Optional[str], Optional[float]]:
    """
    [기능] 현재 호가를 기준가 대비 퍼센트로 환산 후, "범위(구간)"를 판정한다.

    반환:
      (direction, range_key, pct_value)
      - direction: "up"(매도) or "down"(매수) or None
      - range_key: "0.01~0.05" 같은 문자열 또는 ">=1" 또는 None
      - pct_value: 실제 계산된 퍼센트 값(로그 출력용)
    """
    ranges = build_ranges_from_thresholds(thresholds)

    # ✅ 상승(매도) 판정: 매수호가 기준
    if buy_price is not None and base_price > 0:
        pct_up = calc_pct_up(base_price, buy_price)
        if pct_up >= thresholds[0]:
            rk = classify_pct_to_range_key(pct_up, ranges)
            if rk is not None:
                return "up", rk, pct_up

    # ✅ 하락(매수) 판정: 매도호가 기준
    if sell_price is not None and base_price > 0:
        pct_down = calc_pct_down(base_price, sell_price)
        if pct_down >= thresholds[0]:
            rk = classify_pct_to_range_key(pct_down, ranges)
            if rk is not None:
                return "down", rk, pct_down

    return None, None, None


# =========================
# 메인
# =========================
def main() -> None:
    """[기능] 1회 실행 시 조건에 맞으면 '딱 1번' 매수/매도 수행하고 state에 기록"""
    state = load_state()

    base = state.get("base")
    if not isinstance(base, dict) or not base.get("base_price"):
        log_line("AUTO_TRADE", "trade_state.json에 기준가(base_price)가 없습니다. 먼저 kiwoomTradingBuy.py를 실행해 주세요.")
        return

    base_price = int(base["base_price"])
    thresholds = parse_thresholds(base)  # 경계값(0.01/0.05/0.1/0.8/1.0)

    # ✅ 구간별 1회 실행 체크용
    state.setdefault("triggers_done", {})
    state["triggers_done"].setdefault("down", {})
    state["triggers_done"].setdefault("up", {})

    # ✅ 보유수량을 가정으로 관리(실계좌 잔고조회가 아닌 상태)
    state.setdefault("virtual_position", {"stk_cd": TARGET_STK_CD, "stk_nm": TARGET_STK_NM, "qty": 0})
    cur_qty = int(state["virtual_position"].get("qty") or 0)

    cli = KiwoomClient(BASE_URL, APP_KEY, APP_SECRET)
    cli.authenticate()

    buy_price, sell_price = cli.get_best_prices(TARGET_STK_CD)

    # ✅ 퍼센트 계산(로그 표시용)
    pct_up = calc_pct_up(base_price, buy_price) if (buy_price is not None and base_price > 0) else None
    pct_down = calc_pct_down(base_price, sell_price) if (sell_price is not None and base_price > 0) else None

    log_line(
        "AUTO_TRADE",
        f"현재가격 확인 | 종목={TARGET_STK_NM}({TARGET_STK_CD}) | "
        f"매수호가={fmt_won(buy_price)} / 매도호가={fmt_won(sell_price)} | "
        f"기준가={fmt_won(base_price)} | "
        f"상승률(매수호가 기준)={fmt_pct_value(pct_up) if pct_up is not None else 'N/A'} | "
        f"하락률(매도호가 기준)={fmt_pct_value(pct_down) if pct_down is not None else 'N/A'} | "
        f"가정 보유수량={cur_qty}주"
    )

    direction, range_key, pct_value = decide_range_action(base_price, buy_price, sell_price, thresholds)
    if direction is None or range_key is None:
        log_line("AUTO_TRADE", f"대기(HOLD) | 아직 ±{strip_trailing_zeros(thresholds[0])}% 이상 변화가 없습니다.")
        return

    # ✅ band별 중복 실행 방지 (키는 범위 문자열 고정)
    done_map = state["triggers_done"][direction]
    if done_map.get(range_key) is True:
        log_line("AUTO_TRADE", f"대기(HOLD) | 이미 처리한 구간입니다 → {fmt_range_label(direction, range_key)}")
        return

    # ✅ 한 번 실행할 액션(1액션만)
    try:
        if direction == "down":
            od = cli.buy_market(TARGET_STK_CD, TRADE_QTY)
            ord_no = str(od.get("ord_no") or "").strip()
            if not ord_no:
                raise RuntimeError(f"매수 주문 ord_no 없음: {od}")

            log_line(
                "AUTO_BUY",
                f"매수 실행 | {fmt_range_label(direction, range_key)} 도달({fmt_pct_value(pct_value or 0.0)}) → "
                f"시장가 매수 {TRADE_QTY}주 | 주문번호={ord_no}"
            )

            state["virtual_position"]["qty"] = int(cur_qty + TRADE_QTY)
            state["last_order"] = {
                "ts": now_str(),
                "stk_cd": TARGET_STK_CD,
                "stk_nm": TARGET_STK_NM,
                "side": "BUY",
                "ord_no": ord_no,
                "qty_req": TRADE_QTY,
                "price_req": None,
                "trigger": f"DOWN_{range_key}",
                "pct": float(pct_value or 0.0),
            }

        else:
            if cur_qty < TRADE_QTY:
                log_line("AUTO_TRADE", f"매도 불가(HOLD) | 가정 보유수량 부족 ({cur_qty}주 < {TRADE_QTY}주)")
                return

            od = cli.sell_market(TARGET_STK_CD, TRADE_QTY)
            ord_no = str(od.get("ord_no") or "").strip()
            if not ord_no:
                raise RuntimeError(f"매도 주문 ord_no 없음: {od}")

            log_line(
                "AUTO_SELL",
                f"매도 실행 | {fmt_range_label(direction, range_key)} 도달({fmt_pct_value(pct_value or 0.0)}) → "
                f"시장가 매도 {TRADE_QTY}주 | 주문번호={ord_no}"
            )

            state["virtual_position"]["qty"] = int(max(0, cur_qty - TRADE_QTY))
            state["last_order"] = {
                "ts": now_str(),
                "stk_cd": TARGET_STK_CD,
                "stk_nm": TARGET_STK_NM,
                "side": "SELL",
                "ord_no": ord_no,
                "qty_req": TRADE_QTY,
                "price_req": None,
                "trigger": f"UP_{range_key}",
                "pct": float(pct_value or 0.0),
            }

    except KiwoomAPIError as e:
        # ✅ 호출 제한(return_code=5 / 1700)일 때 추가 호출 유도하지 않고 즉시 중단
        if e.return_code == 5 or "1700" in e.return_msg:
            log_line("RATE_LIMIT_HIT", f"허용된 요청 개수 초과로 중단합니다(추가 호출 금지). 상세={e.data}")
            return
        log_line("ORDER_FAILED", f"주문 실패 | code={e.return_code} | msg={e.return_msg} | raw={e.data}")
        raise

    # ✅ “이 구간은 1번 실행했다” 체크
    done_map[range_key] = True
    save_state(state)

    log_line("AUTO_TRADE", f"처리 완료 | {fmt_range_label(direction, range_key)} 은(는) 다시 실행되지 않도록 기록했습니다.")


if __name__ == "__main__":
    main()

3. kiwoomTradingBuy.py 프롬프트

키움 REST 모의투자 쓰는 초보자용 파이썬 스크립트 하나 만들어줘.

삼성전자 005930을 시장가로 100주 1번만 매수하고
매수랑 별개로 내가 넣은 기준가(예: 166200원)를 기준으로
0.01/0.05/0.1/0.8/1% 상하 경계 가격을 계산해서
trade_state.json에 저장해줘.
API 요청 시간, 주문번호, 수량, 기준가 등을 정리해주고
자동매매 금액 구간별로 1회씩만 매수/매도가 실행되도록 해줘.

로그는 trade_log.txt에 한글로 가독성 좋게 남기고, 
로직 실행할 때마다 로그를 시간정보와 함께 내용 남겨줘.

4. kiwoomTradingBuySell.py 프롬프트

kiwoomTradingBuy.py가 만들어준 trade_state.json을 기반으로, 자동매매 스크립트 하나 더 만들어줘.

키움 REST에서 매수호가/매도호가를 가져오고, 
기준가 대비 퍼센트로 변화율을 계산해서 아래 규칙대로 ‘범위 구간’ 으로 판정해줘:

하락: 0.01~0.05 / 0.05~0.1 / 0.1~0.8 / 0.8~1 / 1% 이상 떨어지면 → 20주 시장가 매수
상승: 0.01~0.05 / 0.05~0.1 / 0.1~0.8 / 0.8~1 / 1% 이상 오르면 → 20주 시장가 매도

근데 중요한 건, 실행은 한 번 실행할 때 딱 1액션만 하게 해줘(매수든 매도든 하나만).
그리고 같은 구간은 한 번 처리했으면 다시는 실행하지 않게 json 파일에 기록해줘.
로그는 가독성 있게 한글 문장으로 남겨줘