이 베스트 API C - i beseuteu API C

Archive

eBest xing API

이베스트 xing API

* 이베스트 투자증권 (URL)

- 이베스트 투자증권에서 제공하고 있는 xingAPI는 비교적 쉽게 자동화 시스템을 개발할 수 있는 환경을 제공하고 있다.
- 또한, Python을 이용한 API 호출이 간단한 편에 속한다.

xingAPI

* xingAPI Official Guide Document (URL)
* Package Installation (URL)

- 이베스트 투자증권에서 제공하는 xingAPI는 클라이언트 프로그램에서 사용할 수 있는 API를 제공한다.
- xingAPI는 DLL, COM 버전을 제공하고 있다.
- DLL 방식은 속도가 빠르지만, 사용자 편의성이 COM 방식보다 떨어진다.
- COM 방식은 DLL을 기반으로 실행되지만, 사용자 편의 기능이 포함되어 있다.

- xingAPI는 DevCenter에서 실행한 TR 목록을 DLL이나 COM 방식으로 제공하여
  프로그래밍 방식으로 TR을 호출할 수 있게한다.

COM 방식 (URL)

구성 요소 개체 모델 - Win32 apps

구성 요소 개체 모델

docs.microsoft.com

- 컴포넌트 오브젝트 모델로 Microsoft에서 제공하는 응용프로그램의 Binary Interface이다.
- Python에서 COM을 호출하기 위해선 \(\texttt{pywin32}\) Library가 필요하다.

* \(\texttt{pywin32}\) Installation Instruction

pip install pywin32

XASession Object (XASession 객체)

- 서버 연결과 로그인에 관련된 기능들을 제공하는 COM 객체이다.

Example. Usage of \(\texttt{ConnectServer()}\) Method

>>> import win32com.client
>>> client = win32com.client.Dispatch("XA_Session.XASession")  # COM 타입의 XASession 객체를 불러온다.
>>> client.ConnectServer("demo.ebestsec.co.kr", 20001)         # 이베트스 모의투자 서버 주소, 포트번호
True

- Event들이 발생되었을 때 대처 방법들은 구현이 필요하다.
- Event들은 라이브러리에서 특정 이벤트가 발생하는 시점에 호출되는 Callback-Method 개념으로 정의되어 있다.

Example. Implementation of XASession Events

class XASession:
    #로그인 상태를 확인하기 위한 클래스변수
    login_state = 0

    def OnLogin(self, code, msg):
        """
        로그인 시도 후 호출되는 이벤트.
        code가 0000이면 로그인 성공
        """
        if code == "0000":
            print(code, msg)
            XASession.login_state = 1
        else:
            print(code, msg)

    def OnDisconnect(self):
        """
        서버와 연결이 끊어지면 발생하는 이벤트
        """
        print("Session disconntected")
        XASession.login_state = 0

* \(\texttt{configparser}\) Library (URL)

XAQuery Object (XAQuery 객체)

- TR 조회에 관련된 기능들을 제공하는 COM 객체이다.

XAReal Object (XAReal 객체)

- 실시간 TR을 처리하는 기능들을 제공하는 COM 객체이다.

DevCenter

- API의 테스트를 위해 이베스트에서 제공하는 프로그램이다.
- 각 API에 대한 상세 정보를 확인하고, API를 테스트할 수 있는 환경을 제공한다.
- DevCenter를 처음 설치한 경우 Res 파일 전체를 다운받아야 하는데,
  Res파일이 Transaction(TR)을 요청하는데 필요하기 때문이다.

- TR은 크게 InBlock과 OutBlock으로 구분된다.

InBlock
- TR을 실행하는 데 필요한 값을 의미하고 사용자가 직접 입력해주어야 한다.
  예를 들어, t8436(주식종목조회 API용)에서는 gubun이라는 파라미터가 필요하며
  0(전체), 1(코스피), 2(코스닥) 중에 하나로 입력할 수 있다.

OutBlock
- TR을 실행한 다음 얻을 수 있는 필드에 대한 정보를 나타낸다.
- OutBlock에 Postfix로 OCCURS가 붙어있다면, 이는 데이터가 반복적으로 나오는 구조임을 의미한다.
  (Data Type이 Array임을 의미한다.)

_execute_query() Method

def _execute_query(self, res, in_block_name, out_block_name, *out_fields, **set_fields):
    """
    TR코드를 실행한다.
    이 때, 10분에 200회를 초과하는 TR을 수행시키지 않도록
    TR을 수행할 때 마다 리스트에 TR 수행 시각을 저장하고
    저장된 수행 시각과 현재 시각을 비교해 10분이 넘은 값은
    리스트에서 제거하는 방식으로 수행 시간을 조정한다.
    이 리스트의 길이는 200을 초과하지 않는다.

    [Parameters]
    res            : TR 리소스 이름 (str)
    in_block_name  : 인블록 이름 (str)
    out_block_name : 아웃 블록 이름 (str)
    *out_fields    : 출력필드 리스트 (list)
    **set_fields   : 인블록에 설정할 필드 딕셔너리 (dict)

    [Returns]
    result : 결과 (list)
    """

    time.sleep(1)
    print("current query cnt:", len(self.query_cnt))
    print(res, in_block_name, out_block_name)

    while len(self.query_cnt) >= EBest.QUERY_LIMIT_10MIN:  # 현재 수행된 퀴리의 개수가 200개 이상인 경우
        time.sleep(1)                                      # 프로세스를 1초간 정지
        print("waiting for execute query... current query cnt:", len(self.query_cnt))
        self.query_cnt = list( filter( lambda x: (datetime.today() - x).total_seconds() < EBest.LIMIT_SECONDS, self.query_cnt ) ) # 호출된지 10분 미만인 TR만 추출

    xa_query = win32com.client.DispatchWithEvents("XA_DataSet.XAQuery", XAQuery)  # XAQuery 객체 생성
    xa_query.LoadFromResFile(XAQuery.RES_PATH + res+".res")                       # 리소스 파일 로드

    #in_block_name 셋팅
    for key, value in set_fields.items():
        xa_query.SetFieldData(in_block_name, key, 0, value)
    errorCode = xa_query.Request(0) # TR 요청

    #요청 후 대기
    waiting_cnt = 0
    while xa_query.tr_run_state == 0:  # OnReceiveData 이벤트가 발생되면 tr_run_state는 1로 설정됨 (즉, 0인 경우는 아직 TR 결과가 나오지 않았음을 의미함)
        waiting_cnt +=1
        if waiting_cnt % 1000000 == 0 :
            print("Waiting....", self.xa_session_client.GetLastError())
        pythoncom.PumpWaitingMessages()

    result = []  # 결과 블럭을 담을 리스트
    count = xa_query.GetBlockCount(out_block_name)  # 결과의 개수 계산

    # TR 요청 결과를 result에 저장
    for i in range(count):
        item = {}
        for field in out_fields:  # out_fields Argument에서 정의된 필드값만 추출
            value = xa_query.GetFieldData(out_block_name, field, i)
            item[field] = value
        result.append(item)

    """
    print("IsNext?", xa_query.IsNext)
    while xa_query.IsNext == True:
        time.sleep(1)
        errorCode = xa_query.Request(1)
        print("errorCode", errorCode)
        if errorCode < 0:
            break
        count = xa_query.GetBlockCount(out_block_name)
        print("count", count)
        if count == 0:
            break
        for i in range(count):
            item = {}
            for field in out_fields:
                value = xa_query.GetFieldData(out_block_name, field, i)
                item[field] = value
            print(item)
            result.append(item)
    """

    XAQuery.tr_run_state = 0  # TR 요청 및 결과반환까지 종료
    self.query_cnt.append(datetime.today()) # TR 처리시간 저장

    #영문필드를 한글필드명으로 변환
    for item in result:
        for field in list(item.keys()):
            if getattr(Field, res, None):  # Field 객체에 res 필드를 추출
                res_field = getattr(Field, res, None)
                if out_block_name in res_field:
                    field_hname = res_field[out_block_name]
                    if field in field_hname:
                        item[field_hname[field]] = item[field]  # 한글 필드에 영문 필드의 값 복사
                        item.pop(field)  # 영문 필드명 삭제
    return result

Usage for Transactions

* \(\texttt{gubun}\) Field (시장구분)
"0" (ALL, 모든 종목)
"1" (KOSPI, 코스피 종목)
"2" (KOSDAQ, 코스닥 종목)

Reference: 파이썬과 리액트를 활용한 주식 자동 거래 시스템 구축
(박재현 저, 위키북스, 2020)

Reference: 이베스트투자증권 (URL)

Reference: 이베스트투자증권 xingAPI Official Guide Document (URL)