from __future__ import annotations import httpx from app.core.auth import token_manager from app.core.config import get_base_url, settings from app.core.logger import setup_logger from app.core.rate_limiter import RateLimiter logger = setup_logger("trading") _ORDER_URL = "/uapi/domestic-stock/v1/trading/order-cash" _MODIFY_CANCEL_URL = "/uapi/domestic-stock/v1/trading/order-rvsecncl" _DAILY_CCLED_URL = "/uapi/domestic-stock/v1/trading/inquire-daily-ccld" _TR_IDS = { "real": { "buy": "TTTC0802U", "sell": "TTTC0801U", "modify": "TTTC0803U", "cancel": "TTTC0804U", "daily_ccled": "TTTC0012R", }, "vps": { "buy": "VTTC0802U", "sell": "VTTC0801U", "modify": "VTTC0803U", "cancel": "VTTC0804U", "daily_ccled": "VTTC0012R", }, } class TradingService: def __init__(self) -> None: self._client = httpx.AsyncClient(timeout=10.0) self._rate_limiter = RateLimiter(settings.rate_limit.requests_per_second) def _check_auth(self) -> bool: if not token_manager.is_authenticated: logger.debug("API 인증되지 않음 - 매매 불가") return False return True def _get_tr_id(self, side: str) -> str: mode = settings.kis.server_mode return _TR_IDS.get(mode, _TR_IDS["vps"]).get(side, "VTTC0802U") async def place_order( self, stock_code: str, side: str, qty: int, price: int = 0, order_type: str = "00", ) -> dict: if not self._check_auth(): return {"rt_cd": "-1", "msg": "API 인증되지 않음", "order_no": ""} await self._rate_limiter.acquire() tr_id = self._get_tr_id(side) headers = token_manager.get_auth_headers(tr_id) body = { "CANO": settings.kis.account_no, "ACNT_PRDT_CD": settings.kis.account_code, "PDNO": stock_code, "ORD_DVSN": order_type, "ORD_QTY": str(qty), "ORD_UNPR": str(price) if order_type == "00" else "0", } hashkey = await token_manager.generate_hashkey(body) headers["hashkey"] = hashkey url = f"{get_base_url()}{_ORDER_URL}" resp = await self._client.post(url, json=body, headers=headers) resp.raise_for_status() data = resp.json() result = { "order_no": data.get("output", {}).get("odno", ""), "rt_cd": data.get("rt_cd"), "msg_cd": data.get("msg_cd"), "msg": data.get("msg1"), "stock_code": stock_code, "side": side, "qty": qty, "price": price, "order_type": order_type, } if data.get("rt_cd") == "0": logger.info( "주문 성공: %s %s %s주 %s @ %s원", stock_code, "매수" if side == "buy" else "매도", qty, order_type, price, ) else: logger.warning("주문 실패: %s - %s", stock_code, data.get("msg1")) return result async def modify_order( self, order_no: str, stock_code: str, qty: int, price: int, order_type: str = "00" ) -> dict: if not self._check_auth(): return {"rt_cd": "-1", "msg": "API 인증되지 않음"} await self._rate_limiter.acquire() tr_id = self._get_tr_id("modify") headers = token_manager.get_auth_headers(tr_id) body = { "CANO": settings.kis.account_no, "ACNT_PRDT_CD": settings.kis.account_code, "ODNO": order_no, "PDNO": stock_code, "ORD_DVSN": order_type, "ORD_QTY": str(qty), "ORD_UNPR": str(price), } hashkey = await token_manager.generate_hashkey(body) headers["hashkey"] = hashkey url = f"{get_base_url()}{_MODIFY_CANCEL_URL}" resp = await self._client.post(url, json=body, headers=headers) resp.raise_for_status() data = resp.json() result = { "order_no": data.get("output", {}).get("odno", ""), "rt_cd": data.get("rt_cd"), "msg": data.get("msg1"), } if data.get("rt_cd") == "0": logger.info("정정 성공: 주문번호 %s", order_no) else: logger.warning("정정 실패: %s - %s", order_no, data.get("msg1")) return result async def cancel_order(self, order_no: str, stock_code: str, qty: int) -> dict: if not self._check_auth(): return {"rt_cd": "-1", "msg": "API 인증되지 않음"} await self._rate_limiter.acquire() tr_id = self._get_tr_id("cancel") headers = token_manager.get_auth_headers(tr_id) body = { "CANO": settings.kis.account_no, "ACNT_PRDT_CD": settings.kis.account_code, "ODNO": order_no, "PDNO": stock_code, "ORD_DVSN": "00", "ORD_QTY": str(qty), "ORD_UNPR": "0", } hashkey = await token_manager.generate_hashkey(body) headers["hashkey"] = hashkey url = f"{get_base_url()}{_MODIFY_CANCEL_URL}" resp = await self._client.post(url, json=body, headers=headers) resp.raise_for_status() data = resp.json() result = { "rt_cd": data.get("rt_cd"), "msg": data.get("msg1"), } if data.get("rt_cd") == "0": logger.info("취소 성공: 주문번호 %s", order_no) else: logger.warning("취소 실패: %s - %s", order_no, data.get("msg1")) return result async def get_daily_ccled( self, start_date: str, end_date: str, stock_code: str = "", ccled_dvsn: str = "0", ) -> list[dict]: """KIS API 일일 체결 내역 조회""" if not self._check_auth(): return [] await self._rate_limiter.acquire() tr_id = _TR_IDS.get(settings.kis.server_mode, _TR_IDS["vps"])["daily_ccled"] headers = token_manager.get_auth_headers(tr_id) params = { "CANO": settings.kis.account_no, "ACNT_PRDT_CD": settings.kis.account_code, "FH_PDNO": stock_code, "CCLD_DVSN": ccled_dvsn, "INQR_STRT_DAY": start_date, "INQR_END_DAY": end_date, "WCRC_FRCR_DVSN": "0", "CTAC_TLNO": "", "MKET_ID": "", } url = f"{get_base_url()}{_DAILY_CCLED_URL}" resp = await self._client.get(url, headers=headers, params=params) resp.raise_for_status() data = resp.json() if data.get("rt_cd") != "0": logger.warning("일일 체결 내역 조회 실패: %s", data.get("msg1")) return [] trades = [] for item in data.get("output", []): qty = int(item.get("ft_ccld_qty", 0)) if qty == 0: continue side_code = item.get("sll_buy_dvsn_cd", "") if side_code in ("02", "06"): side = "sell" elif side_code in ("01", "03"): side = "buy" else: side = "buy" if side_code == "01" else "sell" trades.append({ "order_no": item.get("ord_gno_brno", ""), "stock_code": item.get("pdno", ""), "stock_name": item.get("prdt_name", ""), "side": side, "qty": qty, "price": int(float(item.get("ft_ccld_unpr3", 0))), "order_type": item.get("ord_dvsn", "00"), "status": "filled", "order_amount": float(item.get("ft_ord_amt", 0)), "settlement_amount": float(item.get("ft_ccld_amt", 0)), "tax": float(item.get("sttl_evlu_amt", 0)), "commission": float(item.get("ft_lof_ruse_amt", 0)), "created_at": item.get("ord_sttm", ""), "filled_at": item.get("ft_ccld_no", ""), }) logger.info("일일 체결 내역 조회: %d건", len(trades)) return trades async def close(self) -> None: await self._client.aclose() trading_service = TradingService()