ocx_order.py

"""
키움증권 OCX API를 이용한 주문 실행 모듈 (koapy 사용)
"""
import time
import threading
from typing import Dict, Optional, Tuple, Any
import config

from koapy.backend.kiwoom_open_api_plus.core import KiwoomOpenApiPlusEntrypoint


class OCXOrderHandler:
    """OCX API를 이용한 주문 핸들러"""

    def __init__(self):
        self.service: Optional[KiwoomOpenApiPlusEntrypoint] = None
        self.order_result: Optional[str] = None
        self.connected = False

    def connect(self) -> bool:
        """키움증권 OCX API에 연결"""
        try:
            # context manager를 사용하는 경우
            if hasattr(KiwoomOpenApiPlusEntrypoint, '__enter__'):
                self.service = KiwoomOpenApiPlusEntrypoint()
                self.service.__enter__()
            else:
                self.service = KiwoomOpenApiPlusEntrypoint()

            if self.service is None:
                return False

            # EnsureConnected 또는 CommConnect 메서드 확인
            if hasattr(self.service.get_stub(), 'EnsureConnected'):
                self.service.get_stub().EnsureConnected()
            elif hasattr(self.service.get_stub(), 'CommConnect'):
                login_result = self.service.get_stub().CommConnect()
                if login_result != 0:
                    print(f"OCX API 로그인 실패: {login_result}")
                    return False
            elif hasattr(self.service.get_stub(), 'start'):
                self.service.get_stub().start()
                if hasattr(self.service.get_stub(), 'login'):
                    self.service.get_stub().login()

            self.connected = True
            return True
        except Exception as e:
            print(f"OCX API 연결 오류: {e}")
            import traceback
            traceback.print_exc()
            return False

    def ensure_connected(self):
        """연결 확인 및 재연결"""
        if not self.connected or self.service is None:
            if not self.connect():
                raise ConnectionError("OCX API 연결에 실패했습니다.")

    def get_account_number(self) -> Optional[str]:
        """첫 번째 계좌번호 가져오기"""
        try:
            self.ensure_connected()
            if self.service is not None:
                stub = self.service.get_stub()
                account_count = int(stub.GetLoginInfo("ACCOUNT_CNT"))
                if account_count > 0:
                    account_list = stub.GetLoginInfo("ACCNO")
                    accounts = account_list.split(';')
                    return (accounts[0] if accounts
                            else config.ACCOUNT_CONFIG["account_number"])
            return config.ACCOUNT_CONFIG["account_number"]
        except Exception as e:
            print(f"계좌번호 조회 오류: {e}")
            return config.ACCOUNT_CONFIG["account_number"]

    def send_order(
        self,
        code: str,
        quantity: int,
        price: int,
        hoga: str,
        order_type: int
    ) -> Tuple[float, Optional[str]]:
        """
        주문 전송 및 시간 측정

        Returns:
            (주문_전송_시간, 주문번호)
        """
        try:
            self.ensure_connected()
            account_number = self.get_account_number()

            if account_number is None:
                print("계좌번호를 가져올 수 없습니다.")
                return (0.0, None)

            # 주문 전송 시간 측정
            start_time = time.time()

            if self.service is not None:
                stub = self.service.get_stub()
                order_result = stub.SendOrder(
                    "주식주문",  # sRQName
                    "1000",  # sScreenNo
                    account_number,  # sAccNo
                    order_type,  # nOrderType
                    code,  # sCode
                    quantity,  # nQty
                    price,  # nPrice
                    hoga,  # sHogaGb
                    ""  # sOrgOrderNo
                )

            end_time = time.time()
            order_time = end_time - start_time

            self.order_result = str(order_result)
            return (order_time, str(order_result))

        except Exception as e:
            print(f"OCX API 주문 오류: {e}")
            return (0.0, None)



    def execute_order_with_timing(
        self,
        code: Optional[str] = None,
        quantity: Optional[int] = None,
        price: Optional[int] = None,
        hoga: Optional[str] = None,
        order_type: Optional[int] = None
    ) -> Dict[str, Any]:
        """
        주문 실행 및 시간 측정

        Returns:
            {
                'order_time': 주문 전송 시간 (초),
                'order_no': 주문번호
            }
        """
        final_code: str = str(
            code if code is not None else config.ORDER_CONFIG["code"]
        )
        final_quantity: int = int(
            quantity if quantity is not None
            else config.OCX_ORDER_CONFIG["quantity"]
        )
        final_price: int = int(
            price if price is not None else config.ORDER_CONFIG["price"]
        )
        final_hoga: str = str(
            hoga if hoga is not None else config.ORDER_CONFIG["hoga"]
        )
        final_order_type: int = int(
            order_type if order_type is not None
            else config.ORDER_CONFIG["order_type"]
        )

        # 주문 전송
        order_time, order_no = self.send_order(
            final_code, final_quantity, final_price, final_hoga, final_order_type
        )

        return {
            'order_time': order_time,
            'order_no': order_no if order_no else ''
        }

    def disconnect(self):
        """연결 종료"""
        try:
            if self.service:
                self.service.get_stub().CommTerminate()
                self.connected = False
        except Exception as e:
            print(f"OCX API 연결 종료 오류: {e}")


def test_ocx_order():
    """OCX API 주문 테스트"""
    handler = OCXOrderHandler()
    try:
        if handler.connect():
            print("OCX API 연결 성공")
            result = handler.execute_order_with_timing()
            print(f"주문 결과: {result}")
        else:
            print("OCX API 연결 실패")
    finally:
        handler.disconnect()


if __name__ == "__main__":
    test_ocx_order()