6. Trading.py

"""
주문 전송 및 체결 정보 수신에 관한 스크립트

* 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()