Kiwoom REST API

What is it?

Table of Contents

Features

Installation

# install from pypi
pip install -U kiwoom-restful
# install from git fork/clone
pip install -e .

import asyncio, uvloop
uvloop.install()
asyncio.run(main())

Examples

from kiwoom import API, Bot
from kiwoom.http import Response
# API 상세 구현
class MyAPI(API):
 def __init__(self, host:str, appkey: str, secretkey: str):
 super().__init__(host, appkey, secretkey)
 
 # 증권사별매매상위요청
 async def what_stocks_ebest_bought_today(self) -> list[dict]:
 endpoint = '/api/dostk/rkinfo'
 api_id = 'ka10039'
 data = {
 'mmcm_cd': '063', # 회원사코드
 'trde_qty_tp': '0', # 거래량구분 (전체 '0')
 'trde_tp': '1', # 매매구분 (순매수 '1') 
 'dt': '0', # 기간 (당일 '0')
 'stex_tp': '3' # 거래소구분 (통합 '3')
 }
 # 다음 데이터가 있으면 무조건 계속 요청
 should_continue = lambda res: True 
 # 키움 REST API 서버 응답 데이터 취합 후 반환
 res: dict = await self.request_until(
 should_continue, 
 endpoint, 
 api_id, 
 data=data
 )
 key = 'sec_trde_upper'
 if key in res:
 return res[key]
 return [] # or raise Exception
# API 활용 클래스
class MyBot(Bot):
 def __init__(self, host: str, appkey: str, secretkey: str):
 super().__init__(
 host, appkey, secretkey, 
 api=MyAPI(host, appkey, secretkey) 
 )
 self.blacklist: set[str] = set()
 async def add_to_blacklist(self) -> None:
 data = await self.api.what_stocks_ebest_bought_today()
 ETFs = ("KODEX", "TIGER", "RISE")
 top10 = data[:10]
 for item in top10:
 # Roughly skip ETFs
 name = item["stk_nm"]
 if any(etf in name for etf in ETFs):
 continue
 self.blacklist.add(item["stk_cd"])

import asyncio
from kiwoom import Bot
from kiwoom.proc import candle, trade
class MyBot(Bot):
 async def run():
 # 분봉차트 데이터
 code = '005930_AL' # 거래소 통합코드
 df = await self.candle(
 code=code, 
 period='min', # 'tick' | 'min' | 'day'
 ctype='stock', # 'stock' | 'sector'
 start='20250901',
 end='',
 )
 await candle.to_csv(file=f'{code}.csv', path='./', df)
 # 계좌 체결내역 데이터 (최근 2달만)
 fmt = '%Y%m%d'
 today = datetime.today()
 start = today - timedelta(days=60)
 start = start.strftime(fmt)
 end = end.strftime(fmt)
 trs = self.trade(start, end)
 # 키움증권 0343 매매내역 화면 스타일로 저장
 await trade.to_csv('trade.csv', './', trs)

import asyncio
import orjson
import pandas as pd
from kiwoom import Bot
from kiwoom.config.real import RealData
class MyBot(Bot):
 def __init__(self, host: str, appkey: str, secretkey: str):
 super().__init__(host, appkey, secretkey)
 self.ticks: list[dict[str, str]] = []
 async def on_receive_order_book(msg: RealData):
 # 호가 데이터 처리 콜백함수 예시
 # RealData(
 # values: bytes # utf-8 encoded bytes
 # type: str, # API type (ex. 0B, 0D, ...)
 # name: str, # API name (ex. 주식체결, 주식호가잔량, ...)
 # item: str # 종목코드
 # )
 # values는 기호에 맞게 parsing 해서 사용
 # msg.values = json.loads(msg.values) # slow
 msg.values = orjson.loads(msg.values) # fast
 print(
 f"종목코드: {msg.item}, "
 f"최우선매도호가: {msg.values['41']}"
 f"최우선매수호가: {msg.values['51']}"
 )
 
 async def on_receive_tick(msg: RealData):
 # 체결 데이터 처리 콜백함수 예시
 self.list.append(orjson.loads(msg.values))
 if len(self.list) >= 100:
 df = pd.DataFrame(self.list)
 print(df)
 async def run():
 # 거래소 통합 종목코드 받아오기 
 kospi, kosdaq = '0', '10'
 kospi_codes = await bot.stock_list(kospi, ats=True)
 kosdaq_codes = await bot.stock_list(kosdaq, ats=True)
 # 호가 데이터 수신 시 콜백 등록
 self.api.add_callback_on_real_data(
 real_type="0D", # 실시간시세 > 주식호가잔량
 callback=self.on_receive_order_book
 )
 # 체결 데이터 수신 시 콜백 등록
 self.api.add_callback_on_real_data(
 real_tyle="0B", # 실시간시세 > 주식체결
 callback=self.on_receive_tick
 )
 # 데이터 수신을 위한 서버 요청
 # > 데이터가 수신되면 자동으로 콜백이 호출됨
 # > grp_no(그룹 번호) 별 종목코드(최대 100개) 관리 필요
 codes1 = kospi_codes[:100] 
 await self.api.register_hoga(grp_no='1', codes=codes1)
 await self.api.register_tick(grp_no='1', codes=codes1)
 
 # 현재 키움증권 제한으로 100개 코드 지원
 # codes2 = kosdaq_codes[:100]
 # await self.api.register_hoga(grp_no='2', codes=codes2)
 # codes3 = kospi_codes[100:200]
 # await self.api.register_tick(grp_no='3', codes=codes1)
 # 데이터 수신 해제
 await self.api.remove_register(grp_no='1', codes1, type=['0B', '0D']) 
 # await self.api.remove_register(grp_no='2', type='0D') # 호가 '0D'
 # await self.api.remove_register(grp_no='3', type='0B') # 체결 '0B'

import asyncio
from kiwoom import Bot, REAL
async def main():
 # appkey, secretkey 파일 위치 예시
 appkey = "../keys/appkey.txt"
 secretkey = "../keys/secretkey.txt"
 # context 사용을 통한 연결 및 리소스 관리
 async with MyBot(host=REAL, appkey, secretkey) as bot:
 bot.debug(True) # 요청 및 응답 프린트
 await bot.connect()
 await bot.run()
 # context 외부는 자동 연결 해제
 print('Done')
if __name__ == '__main__':
 asyncio.run(main())

Architecture

License

Disclaimer