-
Notifications
You must be signed in to change notification settings - Fork 47
/
6. Trading.py
194 lines (165 loc) · 6.31 KB
/
6. Trading.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
"""
주문 전송 및 체결 정보 수신에 관한 스크립트
* KOA Studio에서 아래 항목 참조
개발가이드 > 주문과 잔고처리 > 기본설명
개발가이드 > 주문과 잔고처리 > SendOrder/SendOrderFO
개발가이드 > 주문과 잔고처리 > OnReceiveChejanData/GetChejanData
개발가이드 > 조회와 실시간데이터처리 > 관련함수 > OnReceiveTrData
* 어떠한 경우에도 손실에 대해서 책임지지 않습니다.
* 일정이 바뻐서 주석을 달지 못했는데 양해부탁드립니다.
"""
import sys
from PyQt5.QtWidgets import QApplication
from PyQt5.QtTest import QTest
from kiwoom import Bot, Server
from kiwoom.data.preps import prep
from kiwoom.utils import name
class Bot(Bot):
def __init__(self, server):
super().__init__(server)
self.acc = ''
self.api.set_connect_hook('on_receive_tr_data', 'rq_name')
self.api.connect('on_receive_tr_data', signal=self.balance, slot=self.server.balance)
self.api.connect('on_receive_tr_data', signal=self.trade, slot=self.server.trade)
self.api.connect('on_receive_chejan_data', slot=self.server.chejan) # without hook
def account(self):
cnt = int(self.api.get_login_info('ACCOUNT_CNT')) # 계좌개수
accounts = self.api.get_login_info('ACCLIST').split(';')[:cnt] # 계좌번호
self.acc = accounts[0]
# Single and Multi Data
def balance(self, prev_next='0'):
tr_code = 'opw00018'
inputs = {
'계좌번호': self.acc,
'비밀번호': '',
'비밀번호입력매체구분': '00',
'조회구분': '1'
}
for key, val in inputs.items():
self.api.set_input_value(key, val)
if prev_next != '0':
QTest.qWait(500)
return_code = self.api.comm_rq_data('balance', tr_code, prev_next, '0000')
if return_code == 0:
self.api.loop()
# Send orders of futures and options
def trade(self):
"""
실계좌일 경우 종목코드 입력하지 마세요.
어떠한 경우에도 손실 책임지지 않습니다.
"""
inputs = (
'trade', # rq_name
'0000', # 화면번호
self.acc, # 계좌번호
'------', # 입력금지!! (종목코드)
1, # 신규매매 (주문종류)
2, # Long (매매구분)
3, # 시장가 (거래구분)
10, # 주문수량
'0', # 주문가격
'', # 원주문번호
)
if self.api.get_login_info('GetServerGubun') != 1:
print('실제 계좌이므로 주문하지 않습니다.')
return
if self.api.send_order_fo(*inputs) == 0:
self.api.loop()
else:
# Do something to handle error
raise RuntimeError(f'Sending order went wrong.')
class Server(Server):
def __init__(self):
super().__init__()
self.downloading = False
def balance(self, scr_no, rq_name, tr_code, record_name, prev_next):
if not self.downloading:
self.downloading = True
keys = ['종목번호', '종목명', '평가손익', '수익률(%)', '보유수량', '매입가', '현재가']
data = {key: list() for key in keys}
cnt = self.api.get_repeat_cnt(tr_code, rq_name)
for i in range(cnt):
for key in keys:
val = prep(self.api.get_comm_data(tr_code, rq_name, i, key))
data[key].append(val)
# Multi Data
for key in keys:
self.share.extend_multi('balance', key, data[key])
if prev_next == '2':
fn = self.api.signal('on_receive_tr_data', 'balance')
fn(prev_next)
else:
# Single Data
for key in ['총평가손익금액', '총수익률(%)']:
val = prep(self.api.get_comm_data(tr_code, rq_name, 0, key))
self.share.update_single(name(), key, val) # name() = 'balance'
self.downloading = False
self.api.unloop()
# Mapped by hook, if rq_name='trade' when on_receive_tr_data() is called.
def trade(self, scr_no, rq_name, tr_code, record_name, prev_next):
num = self.api.get_comm_data(tr_code, rq_name, 0, '주문번호').strip()
if num == '':
self.api.unloop()
raise RuntimeError('Executing order failed.')
# Order filled.
pass
# Mapped directly from on_receive_chejan_data() without hook.
def chejan(self, gubun, item_cnt, fid_list):
# Only for Python >= 3.10.4.
# Use if / elif / else, otherwise.
match gubun:
case '0': # 접수/체결
pass
case '1': # 잔고변경
pass
case '4': # 파생잔고변경
pass
case _:
raise RuntimeError('Execution went wrong.')
# cf. 9203 : 주문번호
for fid in fid_list:
self.api.get_chejan_data(fid)
# Don't forget to self.api.unloop() somewhere.
pass
# 실행 스크립트
if __name__ == '__main__':
app = QApplication(sys.argv)
bot = Bot(Server())
bot.login()
bot.account()
bot.balance()
"""
Check and get single/multi data
* Two different styles.
* Must handle error on your own.
"""
# All single data for balance
if bot.share.isin_single('balance'):
print(bot.share.get_single('balance')) # dict
# Specific single data for balance
if bot.share.isin_single('balance', key='총평가손익금액'):
print(bot.share.get_single('balance', key='총평가손익금액')) # int
# All multi data for balance
try:
bot.share.get_multi('balance') # dict
except KeyError:
pass
# Specific multi data for balance
try:
print(bot.share.get_multi('balance', key='종목명')) # list[str]
except KeyError:
pass
"""
Send orders.
"""
bot.trade()
"""
Interactive mode like jupyter notebook for testing further.
"""
from IPython import embed
print("\nType 'exit()' if you want to quit.\n")
embed()
"""
Keep connection to kiwwom serverr.
"""
# app.exec()