money-expert

선물 실시간 데이터를 받을 일이 생겼습니다. 기존 xing api로 작업을 해서 돌리고 있는데, Open API에 실시간 TR이 있다는 말을 듣고 확인해보았습니다.

websocket을 이용하는 방식이라 개발하기 상당히 쉽게 되어 있습니다. 반면에 xing api는 callback 설정하고 등등 절차가 상당히 복잡합니다. 향후 기능 추가를 생각해서 이번 기회에 Open Api로 바꾸기로 했습니다.

ebest open api 사이트에 가면 실시간 TR 사용법에 대하여 기술되어 있는 글이 있습니다.

https://openapi.ebestsec.co.kr/howto-sample 

전체적인 흐름을 잘 정리해놓았으나 결정적으로 함수를 부르는 main 부분이 없습니다. 

그래서 동작하는 전체 소스를 공유하고자 합니다. Open API를 이용하여 실시간 데이터를 받고자 하는 경우에 유용하게 사용될 것입니다.

import requests
import json
import time
import asyncio
import websockets

APP_KEY = "본인의 key"
SECRET_KEY = "본인의 sec key"

우선 필요한 package import한 후 본인의 Open API 키를 넣습니다.  만약 관련 package가 없다고 나오면 아래와 같이 설치합니다.

pip install websockets

pip install requests

아래 코드는 그냥 묻지도 따지지도 말고 그냥 사용합니다.

# 이하 수정할 필요없는 부분
BASE_URL        = "https://openapi.ebestsec.co.kr:8080"
BASE_URL_WEBS   = "wss://openapi.ebestsec.co.kr:9443/websocket"

   
# 파생인의 쉼터 꿈에님 코드 인용
def get_token():
    PATH = "/oauth2/token"
    headers = {"content-type": "application/x-www-form-urlencoded"}
    body = {
        "appkey": APP_KEY,
        "appsecretkey": SECRET_KEY,
        "grant_type": "client_credentials",
        "scope": "oob"
    }
    result = requests.post(BASE_URL+PATH, headers=headers, data=body)

    body = result.json()

    return body["access_token"]


# 선물 실시간 시세용 message
def reg_future_real(cd, ticker):
    dt = {
        'header': {
            "token": get_token(),
            "tr_type": "3"  #3:실시간 등록
        },  
        'body' : {
            "tr_cd": cd,
            "tr_key": ticker,     
        }         
    }
    return json.dumps(dt) # json을 string으로 변환

지금 만들고자 하는 함수는 실시간 선물 시세입니다. 선물 실시간 체결이 되면 Open API를 통하여 아래 함수로 호출이 옵니다.

# 체결수량 출력
def on_future_sise(data):
    price    = float(data['price'])
    vol      = int(data['cvolume'])
    offerho1 = float(data['offerho1'])
    bidho1   = float(data['bidho1'])

    print(format(price, '5.2f'), format(vol, '4d'), format(offerho1, '5.2f'), format(bidho1, '5.2f'))

실제로 많은 정보들이 data에 들어옵니다. 자세한 정보는 KOSPI200 선물시세를 선택하면 나옵니다.

이제는 on_future_sise() 라는 함수가 호출되는 과정에 대하여 간단하게 정리합니다.

우선 코드부터 보죠.  KOSPI200 선물시세를 받을 수 있는 TR의 CODE는 'FC0'입니다. 

아래 real_api()라는 함수를 보면 async라는 문자가 앞에 있습니다. async with를 사용하는 경우도 있고, await와 함수명을 사용하는 경우도 있습니다. 비동기 함수를 사용하기 위한 절차라고 생각하시고 그냥 무시하셔도 됩니다. 앞으로 real_api 함수에서 connect/send/recv 부분은 손댈필요가 없기 때문입니다.

FUTURE_SISE       = 'FC0'

# 웹 소켓 관련 신경쓸 필요없음
import websockets
async def real_api(real_code, ticker):
    while True:
    # 웹 소켓에 접속을 합니다.
        async with websockets.connect(BASE_URL_WEBS) as websocket:
            str = reg_future_real(real_code, ticker)

            # 웹 소켓 서버로 데이터를 전송합니다.
            await websocket.send(str);
            print('wait')
            time.sleep(2)
            
            while True:
                # 웹 소켓 서버로 부터 메시지가 오면 콘솔에 출력합니다.
                data_s = await websocket.recv();
                data = json.loads(data_s)
                if data['body'] == None: # 시세가 바로 오지 않음 None이면 waiting
                    time.sleep(1)
                    continue
                if real_code == FUTURE_SISE:
                    on_future_sise(data['body'])

그래도 무슨 일을 하는지는 알아야겠지요?

reg_future_real() 함수는 실시간 TR을 등록하는 일을 합니다. 앞으로 이 함수에 등록하는 TR에서 주는 실시간 데이터를 받겠다는 의미입니다. 그 다음부터는 ebest server에서 전달하는 실시간 데이터를 받아서 사용할 함수로 전달하면 끝입니다.

이 중 특이한 부분은 websocket에서 받은 데이터를 json 형태로 변경시키는 아래 부분입니다. 실제로 recv한 data_s는 string 문자열입니다. 긴 문자열로 이루어진 데이터 중 원하는 값을 뽑아내기가 어렵기 때문에 json 형태로 변경해주는 절차가 필요합니다. 그게 바로 json.load입니다. 앞으로 data는 "key":value 형태로 접근이 가능합니다.

                data_s = await websocket.recv();
#                print(data_s)
                data = json.loads(data_s)

여기까지는 ebest sample에 나오는 내용과  비슷합니다.  결정적으로  ebest 문서에는 이렇게 많은 함수를 호출하는 부분이 빠져있는데요.

이렇게 만든 함수를 호출하는 방법은 다음과 같습니다.

if __name__ == '__main__':
    ticker = '101V3000'     # 2024년 3월 만기 선물
    cd = FUTURE_SISE
    asyncio.run(real_api(cd, ticker))

간단하죠.. 그냥 asyncio.run() 함수를 부르면 됩니다. 이때 인자로 위에서 만든 real_api(cd, ticker) 를 넣어줍니다. 즉 비동기적으로 real_api()를 실행하라의 의미입니다.

asyncio.run(real_api(cd, ticker))

결론적으로 websocket으로 연결하고 어쩌고 하는 부분은 모두 모르셔도 됩니다.

실제 선물이 체결되면 아래 함수로 data에 관련 정보가 저장된 상태로 아래 함수가 불리어진다는 것만 인지하면 됩니다.

def on_future_sise(data):

이 함수에서 여러 다양한 일들을 할 수 있습니다. 예를들어 10계약 이상 거래한 것만 출력하자 등등 (이건 다음 글에서 정리해보겠습니다.)

sample 소스코드는 아래 github에 있습니다.

https://github.com/multizone-quant/api_ex/blob/main/open_api_real_sample.py

------------

ebest open api 실시간 TR을 python으로 개발하는 방법을 간단하게 기술하였습니다.

다음에는 이러한 기본 코드를 바탕으로 빌드업을 해보겠습니다.

- 선물 거래량 n개 이상만 출력하기

- 선물 orderbook 실시간 정보 출력하기  

OpenAPI가 나오면서 API 접근성이 많이 개선이 된 것 같습니다.

하지만 처음 자동매매를 시작하시는 분들이 참고할만한 자료가 많지 않아서 진도가 잘 나가지 않을텐데요. 그래서 문서를 보고 직접 코드를 만드는 절차를 자세하게 기술하고자 합니다.

이베스트 증권 OpenAPI를 이용하기 위해서는 API key를 발급받아야합니다. 아래 내용 참고하세요.

API key를 받았다면 아래 API Guide로 이동합니다.

https://openapi.ebestsec.co.kr/apiservice

여기에 보시면 Open API가 지원하는 다양한 TR 정보들에 대한 자세한 정보가 나옵니다. 

"주식/시세/기간별주가" 어떻게 코딩을 하는지 알려드리겠습니다.

API Guide에 나오는 내용입니다.

기본정보에 나오는 값도 코드에 사용을 하여야하니 일단 참고만 합니다.

다음은 요청시에 필요한 header 정보입니다.

뭔가 좀 복잡한데요. t1305를 이용하기 위해서는 header에 이런 정보를 넣어야한다는 의미입니다.

큰 틀은 get_token() 함수를 이용하여 token을 받은 후 본인이 원하는 함수를 호출하는 형식입니다.

token = get_token()

ticker = '069500' # kodex 200
qrycnt = 2          # 몇 개를 받을지
dwmcode = 1         # 일봉:1,  주봉:2, 월봉:3

# infos : candle info
infos = get_stock_dwm_info(ticker, qrycnt, dwmcode)
for info in infos : 
    print(json.dumps(info, indent=4))

일반적으로 t1305를 한번만 부르지 않기 때문에 아래와 같이 함수를 만듭니다. 함수를 만들때에는 인자는 header와 body에 들어갈 정보를 보고 적절하게 만드시면 됩니다. 함수 작성하는 방법 default 값을 assign하는 방법 등은 파이썬 교재를 통하여 습득하시기 바랍니다.

-----------------------

t1305에 해당하는 get_stock_dwm_info()를 만드는 과정을 설명합니다.

header 부분입니다.

PATH는 기본정보의 URL 값을 입력하세요.

연속여부는 edate 값에 따라 결정합니다.

content-type과 authorization에 들어가는 부분은 꿈에님 코드를 활용하였습니다. 이 부분은 예제에 나오는 함수를 그냥 사용하시면 됩니다.

다음으로 각 항목은 tr 마다 모두 틀리기 각 tr에 대하여 guide 참고하여 적절한 값을 넣어야합니다. 만약 api가 작동하지 않는다면 header에 넣어주는 정보가 틀렸다는 의미이므로, guide를 다시 한번 자세하게 읽어서 정확한 값을 넣어주셔야 합니다. 특히 type 부분을 잘 보셔야 합니다. string인지 int인지 자리수는 몇자리인지 잘 확인하셔야 합니다.

def get_stock_dwm_info(ticker, qrycnt, dwmcode=1, edate = ' ', tr_cont_key=''):
    PATH = "/stock/market-data"
    # 연속여부 판단
    cont = 'N'
    if tr_cont_key != '' :
        cont = 'Y'
        
    headers = {
            "content-type": "application/json; charset=utf-8", "authorization": "Bearer "+token,
            "tr_cd": "t1305", 
            "tr_cont": cont, 
            "tr_cont_key": tr_cont_key
    }

다음은 body 부분입니다. header와 마찬가지로 정확한 값을 입력해야합니다. 

date 필드는 처음조회시와 연속조회시 다른 값이 들어갑니다. 일반적으로 한번에 받을 수 있는 자료의 크기가 정해져있으므로, 그 이상 받고 싶은 경우에는 연속조회를 해야합니다. 연속조회 방식은 이전 글 참고하세요.

body에 대응하는 실제 코드입니다. body와 1:1로 위치 맞춰서 적절한 값이 설정되도록 합니다.

    body = {
        "t1305InBlock": { 
            "shcode": ticker, 
            "dwmcode": dwmcode,   # 1@일, 2@주, 3@월
            "date": edate,   # 처음 조회시는 Space 연속 조회시에 이전 조회한 OutBlock의 date 값으로 설정
            "idx": 0,       # 무시
            "cnt": qrycnt   # 건수
            } 
        }

이제 query를 할 준비가 끝났습니다. query를 던집니다.

    result = requests.post(BASE_URL+PATH, headers=headers, data=json.dumps(body))
    
    header = result.headers
    body = result.json()

결과는 header와 body를 통하여 확인이 가능합니다.

header에는 tr 정보가 들어있습니다. 혹시라도 답이 제대로 나오지 않는다면 header 정보를 확인하시면 됩니다. 정상적인 경우에는 무시하셔도 됩니다.

body에는 query한 결과가 들어있습니다. 

body를 보시면 아래 두 key 값으로 결과 값이 저장되어 있습니다. 만약 연속조회를 한다면 "t1305OutBlock"에 있는 date 값을 이용하여 재 조회를 하시면 됩니다. 이렇게 글은 쉽게 썼지만 처음 접하시는 분들은 무슨 말이냐 하실텐데, 이전 선물 tick 받는 소스 참고하여 스스로 공부해보시기 바랍니다.

t1305OutBlock1에는 우리가 원하던 candle 정보가 들어있습니다.

query 결과는 json 형태로 돌려주는데 이 값을 출력할 때  아래 방식을 사용하면 보시기 편할겁니다.

        print(json.dumps(body, indent=4))

마지막으로 함수에서 돌려줄 때 OpenAPI로 받은 값을 그냥 돌려주면 사용하는 쪽에서 힘들어집니다. 앞으로 다양한 API를 이용할텐데 그때마다 t1305OutBlock1 이런 식으로 정해진 값을 key 사용하면 불편해집니다. 따라서 단순하게 dict list로 돌려주는 것이 사용하기 편해집니다. 아래와 같이요.

    return body["t1305OutBlock1"]

이제는 받은 값을 출력해봐야겠죠.

# infos : candle info
infos = get_stock_dwm_info(ticker, qrycnt, dwmcode)
for info in infos : 
    print(json.dumps(info, indent=4))

출력결과입니다.

 잘 나오는군요.

만약 query 하고자 하는 개수가 100개라면 이런식으로 출력하면 너무 양이 많습니다. 시간도 많이 걸리고요. 

이런 경우에는 1-2개만 잘 받는지 확인하고 그 다음부터는 print 문을 빼시면 됩니다.

query한 결과를 화면 출력하였다면 이번에는 파일에 저장하는 방법입니다.

저장하는 방법도 여러가지가 있겠지만 저는 csv 파일 형태로 저장하는 것을 선호합니다.

저장할 때와 읽을 때 사용하는 함수도 포함해놓았으니 필요할 때 사용하시기 바랍니다. 아래는 코드 사용 예입니다.

# 저장    
fname = ticker + '_day.csv'
save_to_file_csv(fname, infos)

# 읽기
data = read_csv_to_dict(fname)
print('읽은 값')
for info in data : 
    print(json.dumps(info, indent=4))

cvs 파일 형태로 저장이 되어 있으므로 excel을 이용하면 뭔가 하기 편합니다.

간단하게 Open API로 본인만의 파이썬 코드를 만드는 방법을 정리해보았습니다. OpenAPI는 많은 TR이 있으므로, 본인이 필요한 TR이 생기면 그때 그때 같은 방식으로 함수를 만들어서 사용하시면 됩니다.

개발하시다가 잘 안되는 경우가 생기면 댓글로 남겨주세요.

소스코드는 아래 github에 있습니다.  OpenAPI_Stock_Candle.py

https://github.com/multizone-quant/api_ex