키움 REST API 조건검색 — CNSRLST로 실시간 편입 받기

키움 REST API 자동매매 완전 가이드에서 토큰·TR·WebSocket을 전반적으로 다뤘지만, 조건검색 하나만 검색해서 들어오는 분이 꾸준합니다. 이유는 분명합니다 — 시세 조회나 주문은 HTTP 요청 한 번으로 끝나는데, 조건검색만 유독 방식이 다르기 때문입니다. API 목록에서 ka10171을 찾아 놓고 "그래서 이걸 어느 주소로 부르지?"에서 막힙니다.

검색식을 만드는 것 자체와 "내가 만든 검색식이 자동 매수가 되는 원리"는 따로 정리한 글이 있습니다. 이 글은 그중 REST API로 구현하는 부분 하나만 다룹니다.

1. 왜 REST 호출이 아니라 WebSocket인가

조건검색은 결과가 한 번에 끝나지 않는 조회이기 때문입니다. 잔고 조회는 물어보면 답이 나오고 끝이지만, 조건검색은 장중 내내 종목이 들어오고 빠집니다. "거래량 급증"이라는 조건이라면 오전 9시 3분에 세 종목, 9시 12분에 두 종목이 더 붙는 식입니다.

이걸 REST로 하려면 봇이 몇 초마다 계속 물어봐야 하고, 그 순간 호출 제한(429)에 부딪힙니다. 그래서 키움은 조건검색을 서버가 밀어 주는 방식으로 설계했고, 그 통로가 WebSocket입니다.

2. 4단계 흐름과 api-id 대응표

문서에서 보이는 ka10171~ka10174는 기능 번호이고, 실제 메시지에 실어 보내는 것은 trnm 값입니다. 이 둘이 따로 놀아서 헷갈립니다.

3. 주고받는 JSON 실물

① 접속 직후 — LOGIN

access_token은 REST 쪽과 동일하게 /oauth2/token에서 받은 값을 씁니다. 조건검색용 별도 토큰이 있는 게 아닙니다.

// 보내는 것
{
  "trnm": "LOGIN",
  "token": "eyJhbGciOi...(access_token)"
}

// 돌아오는 것
{
  "trnm": "LOGIN",
  "return_code": 0,
  "return_msg": ""
}

return_code가 0이면 정상입니다. 0이 아니면 대개 토큰이 만료됐거나 appkey·appsecret으로 받은 토큰이 아닌 경우입니다.

② 조건식 목록 — CNSRLST

// 보내는 것
{ "trnm": "CNSRLST" }

// 돌아오는 것 (형태 예시)
{
  "trnm": "CNSRLST",
  "return_code": 0,
  "return_msg": "",
  "data": [
    ["0", "거래량급증"],
    ["1", "골든크로스"],
    ["2", "52주신고가"]
  ]
}

③ 검색 요청 — CNSRREQ

// 실시간 등록 (search_type "1")
{
  "trnm": "CNSRREQ",
  "seq": "1",             // CNSRLST에서 받은 일련번호
  "search_type": "1",     // "0" = 단발 조회, "1" = 실시간 등록
  "stex_tp": "K",         // 거래소 구분
  "cont_yn": "N",         // 연속조회 여부
  "next_key": ""          // cont_yn 이 "Y"일 때 사용
}

cont_yn·next_key는 결과가 많아 한 번에 안 올 때 쓰는 연속조회 파라미터입니다. 응답에 이어받을 키가 있으면 cont_yn을 "Y"로, next_key에 그 값을 넣어 다시 요청합니다. 조건이 넓어 수백 종목이 잡히는 검색식이라면 첫 응답만 처리하고 끝내는 코드는 종목을 놓칩니다.

4. 파이썬 구현 전문

websockets 라이브러리 기준입니다. 수신 루프의 첫 분기를 PING으로 두는 것이 이 코드의 핵심입니다.

import asyncio, json, websockets

WS_URL = "wss://api.kiwoom.com:10000/api/dostk/websocket"

class ConditionClient:
    def __init__(self, token: str):
        self.token = token
        self.ws = None
        self.conditions = {}      # {조건식이름: seq}
        self.registered = None    # 재접속 시 다시 등록할 seq

    async def connect(self):
        self.ws = await websockets.connect(WS_URL)
        await self.send({"trnm": "LOGIN", "token": self.token})

    async def send(self, payload: dict):
        await self.ws.send(json.dumps(payload))

    async def run(self):
        await self.connect()
        async for raw in self.ws:
            msg = json.loads(raw)
            trnm = msg.get("trnm")

            # ── 1순위: PING을 그대로 되돌려 보낸다 (안 하면 세션 종료) ──
            if trnm == "PING":
                await self.ws.send(raw)
                continue

            if trnm == "LOGIN":
                if msg.get("return_code") != 0:
                    raise RuntimeError(f"로그인 실패: {msg.get('return_msg')}")
                await self.send({"trnm": "CNSRLST"})

            elif trnm == "CNSRLST":
                # data = [[seq, 조건식이름], ...]
                self.conditions = {row[1]: row[0] for row in msg.get("data", [])}
                print("조건식 목록:", self.conditions)
                await self.register("거래량급증")     # 이름으로 지목 (번호 하드코딩 금지)

            elif trnm == "CNSRREQ":
                print("검색 결과 원본:", raw[:500])   # 구조 확인용 — 파서는 이걸 보고 쓴다
                await self.on_result(msg)

            else:
                print("기타 수신:", raw[:300])        # 실시간 push도 여기서 먼저 눈으로 본다

    async def register(self, name: str):
        seq = self.conditions.get(name)
        if seq is None:
            raise KeyError(f"조건식 '{name}' 없음 — HTS에 저장돼 있는지 확인")
        self.registered = seq
        await self.send({
            "trnm": "CNSRREQ", "seq": seq,
            "search_type": "1",        # 실시간
            "stex_tp": "K",
            "cont_yn": "N", "next_key": "",
        })

    async def unregister(self):
        if self.registered:
            await self.send({"trnm": "CNSRCLR", "seq": self.registered})

    async def on_result(self, msg: dict):
        ...   # 6장 참조 — 여기서 바로 주문하지 말 것

async def main(token):
    client = ConditionClient(token)
    while True:                                   # 끊기면 다시 붙고 다시 등록
        try:
            await client.run()
        except Exception as e:
            print("세션 끊김, 5초 후 재접속:", e)
            await asyncio.sleep(5)

5. 자주 막히는 4가지

네 번째는 특히 조건식이 넓을수록 티가 납니다. 백테스트에서는 200종목이 잡히던 조건이 실전에서는 매번 앞쪽 몇십 개만 들어오는 식이라, 전략이 틀린 게 아니라 데이터가 잘린 것인데 원인을 전략에서 찾게 됩니다.

6. 편입 신호를 주문으로 옮길 때

여기가 진짜 어려운 구간입니다. 편입 = 매수로 바로 연결하면 거의 반드시 사고가 납니다.

• 같은 종목이 들락날락합니다. 조건 경계에 걸친 종목은 편입·이탈을 반복해서 보내 줍니다. 그대로 두면 같은 종목을 하루에 여러 번 사고팝니다.
• 장 시작 직후 폭주합니다. 9시 정각에 수십 종목이 한꺼번에 편입되면 주문도 한꺼번에 나가고, 곧바로 호출 제한에 걸립니다.
• 편입 시점과 체결 시점이 다릅니다. 급등으로 조건에 든 종목은 신호와 체결 사이 가격이 벌어지므로 슬리피지를 전제에 넣어야 합니다.

그래서 최소한 이 세 겹은 사이에 둡니다.

bought = set()                 # ① 이미 산 종목은 다시 안 산다
MAX_POSITIONS = 5              # ② 동시 보유 종목 수 상한
queue = asyncio.Queue()        # ③ 신호를 큐에 넣고 정해진 속도로만 주문

async def on_signal(code: str):
    if code in bought or len(bought) >= MAX_POSITIONS:
        return
    bought.add(code)
    await queue.put(code)

async def order_worker():
    while True:
        code = await queue.get()
        # 여기서 키움 주문 REST 호출 (/api/dostk/ordr)
        # 응답의 return_code 를 반드시 확인하고, 실패 시 bought 에서 제거
        await asyncio.sleep(1.0)     # 초당 1건으로 유량 고정

종목코드를 조건검색이 아니라 직접 관리하고 싶다면, 코스피·코스닥 전 종목 목록을 받는 방법은 종목코드 마스터 파일 파싱에 따로 정리해 두었습니다.

자주 묻는 질문

키움 REST API 조건검색은 왜 GET으로 호출이 안 되나요?

조건검색은 결과가 계속 갱신되는 조회라 WebSocket 위에서 동작하도록 설계돼 있습니다. wss://api.kiwoom.com:10000/api/dostk/websocket에 접속한 뒤 trnm이 CNSRLST·CNSRREQ인 JSON을 보내는 방식입니다.

CNSRLST를 먼저 호출해야 하는 이유는?

조건식을 이름이 아니라 일련번호(seq)로 지목하기 때문입니다. 번호는 조건식을 추가·삭제하면 달라질 수 있으므로 하드코딩하지 말고 매번 목록을 받아 이름으로 찾으십시오.

search_type 0과 1은 무엇이 다른가요?

"0"은 지금 조건에 맞는 종목을 한 번 받고 끝, "1"은 실시간 등록이라 편입·이탈이 계속 들어옵니다. 해제는 CNSRCLR입니다.

WebSocket이 몇 분 만에 끊어집니다.

서버가 보내는 trnm: "PING"에 응답하지 않은 경우가 대부분입니다. 받은 메시지를 그대로 되돌려 보내십시오. 그와 별개로 재접속·재등록 로직은 반드시 필요합니다.

조건검색으로 잡힌 종목을 바로 주문에 연결해도 되나요?

권하지 않습니다. 중복 차단·동시 보유 상한·주문 큐 세 겹을 사이에 두십시오. 장 시작 직후 수십 종목이 한꺼번에 편입되는 상황을 전제로 설계해야 합니다.

마무리

LOGIN → CNSRLST → CNSRREQ, 그리고 PING 회신. 조건검색 연동에서 실제로 새로 배워야 하는 건 이 넷뿐입니다. 나머지는 키움 REST API의 일반적인 규칙과 같습니다.

어려운 쪽은 오히려 그 신호를 어떻게 주문으로 바꾸느냐입니다. 검색식은 HTS에서 클릭 몇 번이면 만들어지지만, 그 결과를 중복 없이·감당 가능한 속도로·실패를 실패로 처리하며 주문에 옮기는 부분이 봇 개발의 실제 분량입니다.