-
Notifications
You must be signed in to change notification settings - Fork 0
/
kiwoom.py
203 lines (161 loc) · 7.82 KB
/
kiwoom.py
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
import sys
from PyQt5.QtWidgets import *
from PyQt5.QAxContainer import *
from PyQt5.QtCore import *
import time
import pandas as pd
import sqlite3
class Kiwoom(QAxWidget):
def __init__(self):
super().__init__()
self._create_kiwoom_instance()
self._set_signal_slots()
def _create_kiwoom_instance(self):
self.setControl("KHOPENAPI.KHOpenAPICtrl.1")
def _set_signal_slots(self):
self.OnEventConnect.connect(self._event_connect)
self.OnReceiveTrData.connect(self._receive_tr_data)
self.OnReceiveChejanData.connect(self._receive_chejan_data)
# 계좌 정보
def get_login_info(self, tag):
ret = self.dynamicCall("GetLoginInfo(QString)", tag)
return ret
def comm_connect(self):
self.dynamicCall("CommConnect()")
self.login_event_loop = QEventLoop()
self.login_event_loop.exec_()
def _event_connect(self, err_code):
if err_code == 0:
print("connected")
else:
print("disconnected")
self.login_event_loop.exit()
def get_connect_state(self):
ret = self.dynamicCall("GetConnectState()")
return ret
def set_input_value(self, id, value):
self.dynamicCall("SetInputValue(QString, QString)", id, value)
def comm_rq_data(self, rqname, trcode, next, screen_no):
self.dynamicCall("CommRqData(QString, QString, int, QString)", rqname, trcode, next, screen_no)
self.tr_event_loop = QEventLoop()
self.tr_event_loop.exec_()
def _comm_get_data(self, code, real_type, field_name, index, item_name):
ret = self.dynamicCall("CommGetData(QString, QString, QString, int, QString)", code,
real_type, field_name, index, item_name)
return ret.strip()
#몇개의 데이터가 왔는가
def _get_repeat_cnt(self, trcode, rqname):
ret = self.dynamicCall("GetRepeatCnt(QString, QString)", trcode, rqname)
return ret
def _receive_tr_data(self, screen_no, rqname, trcode, record_name, next, unused1, unused2, unused3, unused4):
if next == '2':
self.remained_data = True
else:
self.remained_data = False
if rqname == "opt10081_req":
self._opt10081(rqname, trcode)
elif rqname == "opw00001_req":
self._opw00001(rqname, trcode)
elif rqname == "opw00018_req":
self._opw00018(rqname, trcode)
try:
self.tr_event_loop.exit()
except AttributeError:
pass
def _opw00001(self, rqname, trcode):
d2_deposit = self._comm_get_data(trcode, "", rqname, 0, "d+2추정예수금")
self.d2_deposit = Kiwoom.change_format(d2_deposit)
def _opt10081(self, rqname, trcode):
data_cnt = self._get_repeat_cnt(trcode, rqname)
for i in range(data_cnt):
date = self._comm_get_data(trcode, "", rqname, i, "일자")
open = self._comm_get_data(trcode, "", rqname, i, "시가")
high = self._comm_get_data(trcode, "", rqname, i, "고가")
low = self._comm_get_data(trcode, "", rqname, i, "저가")
close = self._comm_get_data(trcode, "", rqname, i, "현재가")
volume = self._comm_get_data(trcode, "", rqname, i, "거래량")
self.ohlcv['date'].append(date)
self.ohlcv['open'].append(int(open))
self.ohlcv['high'].append(int(high))
self.ohlcv['low'].append(int(low))
self.ohlcv['close'].append(int(close))
self.ohlcv['volume'].append(int(volume))
def reset_opw00018_output(self):
self.opw00018_output = {'single': [], 'multi': []}
def _opw00018(self, rqname, trcode):
# single data
total_purchase_price = self._comm_get_data(trcode, "", rqname, 0, "총매입금액")
total_eval_price = self._comm_get_data(trcode, "", rqname, 0, "총평가금액")
total_eval_profit_loss_price = self._comm_get_data(trcode, "", rqname, 0, "총평가손익금액")
total_earning_rate = self._comm_get_data(trcode, "", rqname, 0, "총수익률(%)")
estimated_deposit = self._comm_get_data(trcode, "", rqname, 0, "추정예탁자산")
total_earning_rate = Kiwoom.change_format(total_earning_rate)
if self.get_server_gubun():
total_earning_rate = float(total_earning_rate) / 100
total_earning_rate = str(total_earning_rate)
self.opw00018_output['single'].append(Kiwoom.change_format(total_purchase_price))
self.opw00018_output['single'].append(Kiwoom.change_format(total_eval_price))
self.opw00018_output['single'].append(Kiwoom.change_format(total_eval_profit_loss_price))
self.opw00018_output['single'].append(total_earning_rate)
self.opw00018_output['single'].append(Kiwoom.change_format(estimated_deposit))
# multi data
rows = self._get_repeat_cnt(trcode, rqname)
for i in range(rows):
name = self._comm_get_data(trcode, "", rqname, i, "종목명")
quantity = self._comm_get_data(trcode, "", rqname, i, "보유수량")
purchase_price = self._comm_get_data(trcode, "", rqname, i, "매입가")
current_price = self._comm_get_data(trcode, "", rqname, i, "현재가")
eval_profit_loss_price = self._comm_get_data(trcode, "", rqname, i, "평가손익")
earning_rate = self._comm_get_data(trcode, "", rqname, i, "수익률(%)")
quantity = Kiwoom.change_format(quantity)
purchase_price = Kiwoom.change_format(purchase_price)
current_price = Kiwoom.change_format(current_price)
eval_profit_loss_price = Kiwoom.change_format(eval_profit_loss_price)
earning_rate = Kiwoom.change_format2(earning_rate)
self.opw00018_output['multi'].append([name, quantity, purchase_price, current_price,
eval_profit_loss_price, earning_rate])
#주식주문 서버 전송
def send_order(self, rqname, screen_no, acc_no, order_type, code, quantity, price, hoga, order_no):
self.dynamicCall("SendOrder(QString, QString, QString, int, QString, int, int, QString, QString)",
[rqname, screen_no, acc_no, order_type, code, quantity, price, hoga, order_no])
#체결잔고 데이터
def get_chejan_data(self, fid):
ret = self.dynamicCall("GetChejanData(int)", fid)
return ret
#주문체결 데이터
#get_chejan_data 주문체결FID(9203 주문번호, 302 종목명, 900 주문수량, 901 주문가격... 등)
def _receive_chejan_data(self, gubun, item_cnt, fid_list):
print(gubun)
print(self.get_chejan_data(9203))
print(self.get_chejan_data(302))
print(self.get_chejan_data(900))
print(self.get_chejan_data(901))
# 추정예수금 0 제거, 천의자리마다 콤마표시
def change_format(data):
strip_data = data.lstrip('-0')
if strip_data == '':
strip_data = '0'
try:
format_data = format(int(strip_data), ',d')
except:
format_data = format(float(strip_data))
if data.startswith('-'):
format_data = '-' + format_data
return format_data
def change_format2(data):
strip_data = data.lstrip('-0')
if strip_data == '':
strip_data = '0'
if strip_data.startswith('.'):
strip_data = '0' + strip_data
if data.startswith('-'):
strip_data = '-' + strip_data
return strip_data
#서버구분
def get_server_gubun(self):
ret = self.dynamicCall("KOA_Functions(QString, QString)", "GetServerGubun", "")
return ret
def get_code_list_by_market(self, market):
code_list = self.dynamicCall("GetCodeListByMarket(QString)", market)
code_list = code_list.split(';')
return code_list[:-1]