Python, API
[XINGAPI] 해외선물 분봉데이터 조회 - o3123
오늘밤날다
2022. 6. 29. 21:39
o3123InBlock
mktgb에 선물을 조회할 경우 F, 옵션일 경우 O를 입력하고
shcode에 종목코드 그리고 ncnt에 분봉 주기를 정수로 입력하면 된다.
readcnt의 경우에는 출력되는 조회건수를 지정하는 것인데 최대 500건까지 가능하며
cts_data와 cts_time의 경우 연속 조회의 경우에만 사용하기 때문에 일단 최초 조회 시에는 비워두면 된다.
o3123OutBlock / o3123OutBlock1
outblock이 두 가지인데 각각 GetFieldData함수를 통해서 정보를 가져와야 한다.
o3123OutBlock은 분봉 데이터의 가격정보가 담겨있는 o3123OutBlock1의 데이터의 개수와
한 번에 최대 500개까지만 분봉을 조회할 수 있기 때문에
이후 추가적인 데이터를 조회하기 위한 연결고리(?)인 cts_date와 cts_time을 담고 있다.
500개를 넘는 자료가 필요해서 연속 조회를 해야 할 경우에는
최초 o3123InBlock에서 입력한 여러 가지 정보들 중 다른 정보들은 그대로 두고
o3123OutBlock의 cts_date와 cts_time만 다시 입력해서 server로 보낸다.
이후 새로운 데이터를 담은 o3123OutBlock / o3123OutBlock1이 넘어온다고 생각하면 된다.
이 과정을 반복하면 필요한 과기처의 정보를 연속해서 조회할 수 있다.
o3123OutBlock의 데이터 예시가 아래와 같을 때 아래의 연속 일자와 연속 시간이 연속조회에 필요한 정보 값이 된다.
| {'단축코드': 'HMHN22', '시차': '-1', '조회건수': '1', '연속일자': '20220629', '연속시간': '203000'} |
위 내용을 전체 코드로 구성하면 아래와 같다.
미니항셍 연결선물의 5분봉을 최대 개수까지(500개)(500개) 조회하고
다시 연속 조회를 한 번만 추가해서 총 1,000개의 5분봉 데이터를 딕셔너리로 저장한 후,
이를 데이터프레임으로 변환한 후 csv파일로 저장한다.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
import win32com.client
import pythoncom
import pandas as pd
import time
pd.set_option('display.max_colwidth', 0)
# 로그인 이벤트 클래스
class XASessionEventHandler:
login_state = 0
def OnLogin(self, code, msg):
if code == "0000":
XASessionEventHandler.login_state = 1
# TR 이벤트 클래스
class XAQueryEventHandlerTR:
query_state = 0
def OnReceiveData(self, code):
XAQueryEventHandlerTR.query_state = 1
class Main:
def __init__(self):
# TRHandler 등록
self.instXAQueryTR = win32com.client.DispatchWithEvents("XA_DataSet.XAQuery", XAQueryEventHandlerTR)
# 조회횟수 변수선언
self.req_count = 0
# 로그인 함수
def connect(self):
id = ""
passwd = ""
cert_passwd = ""
instXASession = win32com.client.DispatchWithEvents("XA_Session.XASession", XASessionEventHandler)
instXASession.ConnectServer("demo.ebestsec.co.kr", 20001)
instXASession.Login(id, passwd, cert_passwd, 0, 0)
while XASessionEventHandler.login_state == 0:
pythoncom.PumpWaitingMessages()
# TR.Request 함수
def tr_request(self, req_con):
time.sleep(0.2)
XAQueryEventHandlerTR.query_state = 0
self.instXAQueryTR.Request(req_con)
while self.instXAQueryTR.query_state == 0:
pythoncom.PumpWaitingMessages()
self.req_count = self.req_count + 1
# 해외선물 옵션차트(분)
def req_o3123(self, mktgb, shcode, ncnt, readcnt, cts_date, cts_time):
# outblock 정의
outblock = {'단축코드': 'shcode', '시차': 'timediff', '조회건수': 'readcnt', '연속일자': 'cts_date', '연속시간': 'cts_time'}
outblock1 = {'날짜': 'date', '현지시각': 'time', '시가': 'open', '고가': 'high', '저가': 'low', '종가': 'close',
'거래량': 'volume'}
# 1. res파일 지정
self.instXAQueryTR.ResFileName = 'C:/eBEST/xingAPI/Res/o3123.res'
# 2. 데이터 입력
self.instXAQueryTR.SetFieldData("o3123InBlock", "mktgb", 0, mktgb)
self.instXAQueryTR.SetFieldData("o3123InBlock", "shcode", 0, shcode)
self.instXAQueryTR.SetFieldData("o3123InBlock", "ncnt", 0, ncnt)
self.instXAQueryTR.SetFieldData("o3123InBlock", "readcnt", 0, readcnt)
self.instXAQueryTR.SetFieldData("o3123InBlock", "cts_date", 0, cts_date)
self.instXAQueryTR.SetFieldData("o3123InBlock", "cts_time", 0, cts_time)
# 3. 서버전송 - 최초 조회시 0
self.tr_request(0)
# 4. OutBlock 데이터 가져오기 (조회건수, cts_date, cts_time 등)
ret_outblock = {}
for item in outblock:
ret_outblock[item] = self.instXAQueryTR.GetFieldData("o3123OutBlock", outblock[item], 0)
print(ret_outblock)
# OutBlock 중 조회건수를 정수로 변환 후 item_count변수로 지정
item_count = int(ret_outblock['조회건수'])
# cts변수 업데이트
cts_date = ret_outblock['연속일자']
cts_time = ret_outblock['연속시간']
if cts_date == '':
con_req = False
else:
con_req = True
# 4.1 OutBlock1 데이터 가져오기 (분봉데이터)
result = {}
for i in range(item_count): # 조회건수 만큼 반복하여 조회
temp_dict = {}
for item in outblock1:
temp_dict[item] = self.instXAQueryTR.GetFieldData("o3123OutBlock1", outblock1[item], i)
result[len(result)] = temp_dict
# 연속조회 시작 -------------------------------------------------------------------------------------------------
while con_req:
print(f'연속조회시작: {cts_date}, {cts_time}')
# 2. 데이터 입력
self.instXAQueryTR.SetFieldData("o3123InBlock", "cts_date", 0, cts_date)
self.instXAQueryTR.SetFieldData("o3123InBlock", "cts_time", 0, cts_time)
# 3. 서버전송 - 연속조회일 경우 1
self.tr_request(1)
# 4. OutBlock 데이터 가져오기 (조회건수, cts_date, cts_time 등)
ret_outblock = {}
for item in outblock:
ret_outblock[item] = self.instXAQueryTR.GetFieldData("o3123OutBlock", outblock[item], 0)
print(ret_outblock)
# OutBlock 중 조회건수를 정수로 변환 후 item_count변수로 지정
item_count = int(ret_outblock['조회건수'])
# cts변수 업데이트
cts_date = ret_outblock['연속일자']
cts_time = ret_outblock['연속시간']
if cts_date == '': # cts_data가 비어있을 경우 연속조회 중지를 위해 con_req를 False로 바꿈
con_req = False
else:
con_req = True
# 4.1 OutBlock1 데이터 가져오기 (분봉데이터)
for i in range(item_count): # 조회건수 만큼 반복하여 조회
temp_dict = {}
for item in outblock1:
temp_dict[item] = self.instXAQueryTR.GetFieldData("o3123OutBlock1", outblock1[item], i)
result[len(result)] = temp_dict
# 조회횟수 2번째에 중지 - o3123 조회제한: 10분당 200건
if self.req_count == 2:
break
# 최종 dictionary 결과물과 dictionary를 dataframe으로 변환한 결과물 출력
print('조회데이터 출력 ----------------------------------------------------------------------------------------')
print(result)
df = pd.DataFrame(result).transpose()
print('DataFrame 출력 ---------------------------------------------------------------------------------------')
print(df)
df.to_csv('hmh_minutes_ohlcv.csv')
if __name__ == "__main__":
main = Main()
main.connect()
# 분봉업데이트
mktgb = 'F'
shcode = 'HMH.1'
ncnt = 5
readcnt = 500
cts_date = ''
cts_time = ''
main.req_o3123(mktgb, shcode, ncnt, readcnt, cts_date, cts_time)
|
cs |
