Simple Python Wrapper for Kiwoom RESTful API
- 키움증권에서 제공하는 'REST' API 인터페이스 사용을 위한 간단한 Python Wrapper 모듈
- 네트워크 프로토콜 디테일은 숨기고 투자 전략 로직에 집중할 수 있도록 설계
- 직접 개발하고자 하는 사람을 위한 모듈로 부가적인 기능은 최대한 배제
- 'REST' API가 아닌 기존 PyQt 기반 OCX 방식 API는 이곳을 참조
- 간결한 API를 통한 빠른 프로토 타입 및 전략 최적화
- Async Http 요청 및 응답 + Websocket 데이터 처리
- 실시간 데이터 처리를 위한 non-블로킹 콜백 시스템
- msgpsec / orjson 기반 실시간 고속 json 파싱
- 초당 Http 연결/호출 제한 자동관리
- Websocket ping-pong 자동처리
모듈 관련 상세한 API 문서 페이지는 이곳을 참고해 주세요.
API에 변화가 없다면 구조상 큰 변화는 없을 예정입니다. (RC 단계)
# install from pypi
pip install -U kiwoom-restful
# install from git fork/clone
pip install -e .Requirements
- python 3.11+ 권장 (최소 3.10+)
- 리눅스 환경에서는 uvloop 추가설치로 속도 향상 가능
import asyncio, uvloop
uvloop.install()
asyncio.run(main())기본적인 API 구현 및 Bot 활용예시
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"])Bot 기본 내장함수 2가지 활용예시
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)Websocket 실시간 데이터 요청 구현예시
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())Layered Roles
-
Client : Http 요청 횟수 제한 관리 및 데이터 연속 조회 관리
-
Socket : WebSocket 연결 및 수명 관리, 데이터 수신 후 asyncio.Queue에 전달
-
API :
- 기본적인 REST API Http 요청 및 응답 관리
- Queue를 소비하며 ping-pong, login 및 데이터 디코딩 수행
- 실시간 데이터 주제별 스트림/콜백/구독 관리
-
Bot : API 기능을 활용하여 전략을 수행하도록 사용자가 구현
MIT License © Contributors
- 본 프로젝트는 키움증권 공식 프로젝트가 아닙니다.
- 실제 운용 전 모의 테스트 환경에서 충분히 검증 바랍니다.
- 발생한 어떠한 손실에 대하여도 본 프로젝트의 개발자는 책임이 없음을 알립니다.