Xing API 모의투자 - Xing API mouituja

퀀트투자를 위한 xing API 파이썬 레시피 01. 개발환경 준비하기 01) 아나콘다 설치하기 02) PyCharm 설치하기 02. xingAPI 사용 준비하기 01) 비대면 계좌개설 02) API 사용등록 및 설치 03. xingAPI 기초 01) 로그인 02) 계좌 가져오기 03) DevCenter 사용하기 04) 단일 데이터 조회 04. COM과 스레드 01) COM 기초 02) threading 모듈 03) COM과 스레드 04) COM 아파트먼트 모델 05) queue 사용하기 05. pyxing 01) pyxing 설치하기 02) pyxing으로 로그인 03) pyxing으로 TR 요청하기 04) pyxing으로 Real 요청하기 06. 주문 01) 계좌 이체 및 모의투자 신청하기 09. DDE 01) DDE 사용해보기 10. TR 목록 01) 주식 t8430 주식종목조회 20. 부록 1) 클래스 문법

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)