import asyncio
import datetime as dtm
import math
from collections import defaultdict
from decimal import Decimal
from typing import AsyncGenerator, Dict, List, Optional
import pandas as pd
import requests
from pyqqq.brokerage.toss.domestic_stock import TossDomesticStock
from pyqqq.brokerage.toss.oauth import TossAuth
from pyqqq.datatypes import *
from pyqqq.utils.logger import get_logger
from pyqqq.utils.market_schedule import get_market_schedule, is_full_day_closed
[docs]
class TossStockPosition(StockPosition):
pass
[docs]
class TossStockOrder(StockOrder):
pass
# Toss 주문 status 중 "진행 중"으로 취급하는 상태 (OPEN 그룹)
_PENDING_STATUSES = {"PENDING", "PARTIAL_FILLED", "PENDING_CANCEL", "PENDING_REPLACE"}
# 한 번에 조회 가능한 최대 심볼 수 (Toss /prices 제한)
_MAX_SYMBOLS_PER_REQUEST = 200
# 페이지네이션 루프 안전 상한
_MAX_PAGES = 100
# 호가 레벨 수 - 10호가 미만 응답 시 0 으로 채움 (KIS 와 동일한 형태)
_ORDERBOOK_DEPTH = 10
def _error_code(e: requests.HTTPError) -> Optional[str]:
"""HTTPError 응답 본문에서 에러 코드(``error.code``)를 추출한다. 파싱 불가 시 None."""
try:
return e.response.json()["error"]["code"]
except Exception:
return None
[docs]
class TossSimpleDomesticStock:
"""
토스증권 국내 주식 API를 사용하여 주식 거래를 하기 위한 클래스입니다.
기존 TossDomesticStock 클래스를 감싸고, 간단한 주문/조회 기능을 제공합니다.
메서드 시그니처는 :class:`KISSimpleDomesticStock` 을 따릅니다.
토스증권 Open API 제약으로 인한 차이점:
- 거래대금(value) 데이터가 제공되지 않아 관련 값은 ``None``/``NA`` 로 반환됩니다.
- 주문의 체결 거래소(KRX/NXT/SOR)가 노출되지 않아 ``StockOrder.exchange`` 는 항상 ``None``
이며, ``exchanges`` 필터 인자는 받되 무시됩니다.
- 실시간 주문 이벤트(웹소켓)가 아직 지원되지 않아 ``listen_order_event`` 는
``NotImplementedError`` 를 발생시킵니다. (TradingTracker 사용 불가)
- 정정 주문 시 새로운 주문 ID가 발급되고 원주문은 ``REPLACED`` 상태가 됩니다.
Args:
auth (TossAuth): 인증 정보
account_no (str): 계좌번호. 미지정 시 첫 번째 계좌를 사용한다. (하나의 App Key로 여러 계좌 지원)
"""
[docs]
def __init__(self, auth: TossAuth, account_no: Optional[str] = None):
self.auth = auth
self.stock_api = TossDomesticStock(auth)
self._account_no = account_no
self.account_product_code = "" # tracker 호환용
self.logger = get_logger(__name__ + ".TossSimpleDomesticStock")
@property
def account_no(self) -> str:
if self._account_no is None:
accounts = self.stock_api.get_accounts()
if not accounts:
raise ValueError("사용 가능한 계좌가 없습니다.")
self._account_no = accounts[0]["accountNo"]
return self._account_no
# ------------------------------------------------------------------ #
# 계좌 / 자산
# ------------------------------------------------------------------ #
[docs]
def get_account(self) -> dict:
"""
계좌 정보를 조회합니다.
Returns:
dict: 계좌 정보
- total_balance (int): 총 평가 금액 (주식 평가 금액 + 매수 가능 현금)
- purchase_amount (int): 매입 금액
- evaluated_amount (int): 평가 금액
- pnl_amount (int): 평가 손익
- pnl_rate (Decimal): 손익률
"""
holdings = self.stock_api.get_holdings(self.account_no)
buying_power = self.stock_api.get_buying_power(self.account_no, "KRW")
purchase_amount = int(holdings["totalPurchaseAmount"]["krw"])
evaluated_amount = int(holdings["marketValue"]["amount"]["krw"])
pnl_amount = int(holdings["profitLoss"]["amount"]["krw"])
pnl_rate = Decimal(pnl_amount / purchase_amount) * 100 if purchase_amount != 0 else Decimal(0)
investable_cash = int(buying_power["cashBuyingPower"])
return {
"total_balance": evaluated_amount + investable_cash,
"purchase_amount": purchase_amount,
"evaluated_amount": evaluated_amount,
"pnl_amount": pnl_amount,
"pnl_rate": pnl_rate,
}
[docs]
def get_possible_quantity(self, asset_code: str, order_type: OrderType = OrderType.MARKET, price: int = 0) -> dict:
"""
주문 가능 수량을 조회합니다.
수량은 매매 수수료(원 단위 올림)를 포함한 총비용이 매수 가능 현금을 넘지 않는
최대치로 계산합니다.
Args:
asset_code (str): 종목 코드
order_type (OrderType): 주문 유형 (LIMIT/MARKET)
price (int): 주문 가격. 지정가 주문 시 필수이며, 시장가 주문에서는 무시된다.
Returns:
dict: 주문 가능 수량 정보
- investable_cash (int): 주문 가능 현금
- reusable_amount (int): 재사용 가능 금액 (제공되지 않음 - 항상 0)
- price (int): 계산 기준 단가 (지정가 또는 상한가)
- quantity (int): 주문 가능 수량 (수수료 반영)
- amount (int): 주문 시 소요 금액 (수수료 제외 주문 금액)
"""
if order_type not in (OrderType.LIMIT, OrderType.MARKET):
raise ValueError("지원하지 않는 주문 유형입니다.")
buying_power = self.stock_api.get_buying_power(self.account_no, "KRW")
investable_cash = int(buying_power["cashBuyingPower"])
if order_type == OrderType.LIMIT:
if int(price) <= 0:
raise ValueError("지정가 주문은 가격이 필요합니다.")
unit_price = int(price)
else:
# 시장가는 상한가 기준 - 체결가와 무관하게 주문 가능한 보수적 수량
limits = self.stock_api.get_price_limits(asset_code)
upper = limits.get("upperLimitPrice")
if upper is not None:
unit_price = int(upper)
else:
# 상한가 미제공(가격제한 없는 시장) 방어 - 현재가로 폴백
prices = self.stock_api.get_prices([asset_code])
unit_price = int(prices[0]["lastPrice"]) if prices else 0
# FIXME 국내 수수료율이 부정확
# commission_rate = self.stock_api._commission_rate(self.account_no)
commission_rate = Decimal(0.00015)
# 수수료 포함 총비용(주문 금액 + 올림 처리한 수수료)이 현금을 넘지 않는 최대 수량
quantity = int(Decimal(investable_cash) / (Decimal(unit_price) * (1 + commission_rate))) if unit_price > 0 else 0
while quantity > 0 and quantity * unit_price + math.ceil(quantity * unit_price * commission_rate) > investable_cash:
quantity -= 1
return {
"investable_cash": investable_cash,
"reusable_amount": 0,
"price": unit_price,
"quantity": quantity,
"amount": quantity * unit_price,
}
[docs]
def get_positions(self) -> List[StockPosition]:
"""
보유 종목을 조회합니다. (국내 주식만)
Note:
보유 주식 응답에는 매도 가능 수량이 없어, 보유 수량에서 미체결 매도 주문
수량을 차감하여 계산합니다. 담보/대여 등으로 잠긴 수량은 반영되지 않으므로
실제보다 클 수 있습니다.
Returns:
List[StockPosition]: 보유 종목 정보 리스트
"""
holdings = self.stock_api.get_holdings(self.account_no)
items = [item for item in (holdings.get("items") or []) if int(item["quantity"]) > 0]
if not items:
return []
pending_sell = defaultdict(int)
open_orders = self.stock_api.get_orders(self.account_no, {"status": "OPEN"})
for order in open_orders.get("orders") or []:
if order.get("side") == "SELL":
quantity = int(order.get("quantity") or 0)
filled = int((order.get("execution") or {}).get("filledQuantity") or 0)
pending_sell[order["symbol"]] += max(0, quantity - filled)
return [self._to_stock_position(item, pending_sell.get(item["symbol"], 0)) for item in items]
@staticmethod
def _to_stock_position(item: dict, pending_sell_quantity: int = 0) -> StockPosition:
"""보유 종목 응답 항목을 StockPosition 으로 변환한다."""
quantity = int(item["quantity"])
return StockPosition(
asset_code=item["symbol"],
asset_name=item.get("name", ""),
quantity=quantity,
sell_possible_quantity=max(0, quantity - pending_sell_quantity),
average_purchase_price=Decimal(item["averagePurchasePrice"]),
current_price=int(item["lastPrice"]),
current_value=int(item["marketValue"]["amount"]),
current_pnl=Decimal(item["profitLoss"]["rate"]) * 100, # 소수비율(0.1077=10.77%)로 내려와 % 단위로 변환
current_pnl_value=int(item["profitLoss"]["amount"]),
)
# ------------------------------------------------------------------ #
# 시세
# ------------------------------------------------------------------ #
[docs]
def get_price(self, asset_code: str, data_exchange: DataExchange = DataExchange.KRX) -> dict:
"""
주식 현재 가격 조회
Note:
거래대금이 제공되지 않아 ``value`` 는 항상 ``None`` 입니다.
당일 캔들이 아직 없으면 시가/고가/저가/거래량은 0 입니다.
Args:
asset_code (str): 종목코드
data_exchange (DataExchange): 미지원. 통합(KRX+NXT) 시세만 제공됨
Returns:
dict: 현재 가격 정보
- code (str): 종목 코드
- current_price (int): 현재 가격
- volume (int): 거래량
- open_price (int): 시가
- high_price (int): 고가
- low_price (int): 저가
- max_price (int|None): 상한가
- min_price (int|None): 하한가
- diff (int): 전일대비
- diff_rate (float): 전일대비율
- value (None): 거래대금 (제공되지 않음)
"""
prices = self.stock_api.get_prices([asset_code])
current_price = int(prices[0]["lastPrice"]) if prices else 0
snapshot = self._daily_candle_snapshot(asset_code, current_price)
limits = self.stock_api.get_price_limits(asset_code)
upper = limits.get("upperLimitPrice")
lower = limits.get("lowerLimitPrice")
return {
"code": asset_code,
"current_price": current_price,
"volume": snapshot["volume"],
"open_price": snapshot["open_price"],
"high_price": snapshot["high_price"],
"low_price": snapshot["low_price"],
"max_price": int(upper) if upper is not None else None,
"min_price": int(lower) if lower is not None else None,
"diff": snapshot["diff"],
"diff_rate": snapshot["diff_rate"],
"value": None,
}
def _daily_candle_snapshot(self, asset_code: str, current_price: int) -> dict:
"""당일 일봉(시가/고가/저가/거래량)과 전일 종가 대비 등락을 조회한다.
당일 캔들이 아직 없으면 시가/고가/저가/거래량은 0 이며, 등락은 가장 최근
캔들의 종가를 전일 종가로 사용해 계산한다.
"""
candle_page = self.stock_api.get_candles(asset_code, interval="1d", count=2)
candles = candle_page.get("candles") or []
today = dtm.date.today()
today_candle = candles[0] if candles and candles[0]["timestamp"].date() == today else None
if today_candle is not None:
prev_candle = candles[1] if len(candles) > 1 else None
else:
prev_candle = candles[0] if candles else None
prev_close = int(prev_candle["closePrice"]) if prev_candle else 0
diff = current_price - prev_close if prev_close else 0
diff_rate = float(diff / prev_close * 100) if prev_close else 0.0
return {
"volume": int(today_candle["volume"]) if today_candle else 0,
"open_price": int(today_candle["openPrice"]) if today_candle else 0,
"high_price": int(today_candle["highPrice"]) if today_candle else 0,
"low_price": int(today_candle["lowPrice"]) if today_candle else 0,
"diff": diff,
"diff_rate": diff_rate,
}
[docs]
def get_price_for_multiple_stock(self, asset_codes: List[str]) -> pd.DataFrame:
"""
여러 종목의 현재 가격 조회
Note:
현재가는 다건 조회(200개 단위)로 가져오지만, 시가/고가/저가/거래량/등락은
현재가 응답에 없어 종목마다 일봉 캔들을 조회해 채웁니다.
(종목 수만큼 캔들 API 호출 발생 - MARKET_DATA_CHART 5회/초 제한 적용)
Args:
asset_codes (List[str]): 종목 코드 리스트
Returns:
pd.DataFrame: 현재 가격 정보 (index: code)
"""
price_map = self._fetch_current_prices(asset_codes)
rows = []
for code in asset_codes:
if code not in price_map:
continue
current_price = price_map[code]
snapshot = self._daily_candle_snapshot(code, current_price)
rows.append(
{
"code": code,
"current_price": current_price,
"volume": snapshot["volume"],
"open_price": snapshot["open_price"],
"high_price": snapshot["high_price"],
"low_price": snapshot["low_price"],
"diff": snapshot["diff"],
"diff_rate": snapshot["diff_rate"],
}
)
df = pd.DataFrame(rows, columns=["code", "current_price", "volume", "open_price", "high_price", "low_price", "diff", "diff_rate"])
df.set_index("code", inplace=True)
return df
def _iter_candles(self, asset_code: str, interval: str, until_date: dtm.date, adjusted: bool = True):
"""캔들을 최신부터 과거로 페이지네이션하며 순회한다. until_date 이전 캔들을 만나면 종료."""
before = None
for _ in range(_MAX_PAGES):
page = self.stock_api.get_candles(asset_code, interval=interval, count=200, before=before, adjusted=adjusted)
candles = page.get("candles") or []
if not candles:
return
for candle in candles:
if candle["timestamp"].date() < until_date:
return
yield candle
next_before = page.get("nextBefore")
if next_before is None or next_before == before:
return
before = next_before
[docs]
def get_historical_daily_data(self, asset_code: str, first_date: dtm.date, last_date: dtm.date, adjusted_price: bool = True, data_exchange: DataExchange = DataExchange.KRX) -> pd.DataFrame:
"""
일봉 데이터 검색
Note:
거래대금이 제공되지 않아 ``value`` 컬럼은 ``NA`` 입니다.
조회 기간 중 데이터가 없는 영업일은 0 으로 채웁니다.
Args:
asset_code (str): 종목코드
first_date (datetime.date): 조회 시작일자
last_date (datetime.date): 조회 종료일자
adjusted_price (bool): 수정 주가 여부
data_exchange (DataExchange): 미지원. 통합(KRX+NXT) 시세만 제공됨
Returns:
pd.DataFrame: 일봉 데이터 (시간의 역순)
"""
assert first_date <= last_date, "last_date는 first_date와 같거나, 이후 날짜여야 합니다"
assert last_date <= dtm.date.today(), "last_date는 오늘과 같거나 이전이어야 합니다."
rows = []
for candle in self._iter_candles(asset_code, "1d", first_date, adjusted=adjusted_price):
date = candle["timestamp"].date()
if date > last_date:
continue
rows.append(
{
"date": date,
"open": int(candle["openPrice"]),
"high": int(candle["highPrice"]),
"low": int(candle["lowPrice"]),
"close": int(candle["closePrice"]),
"volume": int(candle["volume"]),
"value": pd.NA,
}
)
df = pd.DataFrame(rows, columns=["date", "open", "high", "low", "close", "volume", "value"])
df["date"] = pd.to_datetime(df["date"])
df.set_index("date", inplace=True)
# 조회기간 중 데이터가 없는 날짜는 0 으로 채움
dates = pd.date_range(start=first_date, end=last_date, freq="B")
if len(dates) != len(df):
dates = [date for date in dates if not is_full_day_closed(date)]
df = df.reindex(dates, fill_value=0)
df = df.sort_index(ascending=False)
return df
[docs]
def get_today_minute_data(self, asset_code: str, data_exchange: DataExchange = DataExchange.KRX) -> pd.DataFrame:
"""
분봉 데이터 검색
Note:
거래대금이 제공되지 않아 ``value``/``cum_value`` 컬럼은 ``NA`` 입니다.
휴장일에는 빈 DataFrame을 반환합니다.
Args:
asset_code (str): 종목코드
data_exchange (DataExchange): 미지원. 통합(KRX+NXT) 시세만 제공됨
Returns:
pd.DataFrame: 분봉 데이터 (시간의 역순)
"""
def _create_minute_dataframe(data: Optional[list] = None) -> pd.DataFrame:
df = pd.DataFrame(data or [], columns=["time", "open", "high", "low", "close", "volume", "value", "cum_volume", "cum_value"])
df["time"] = pd.to_datetime(df["time"])
df.set_index("time", inplace=True)
return df
today = dtm.date.today()
schedule = get_market_schedule(today)
if schedule.full_day_closed:
return _create_minute_dataframe()
rows = []
for candle in self._iter_candles(asset_code, "1m", today):
rows.append(
{
"time": candle["timestamp"],
"open": int(candle["openPrice"]),
"high": int(candle["highPrice"]),
"low": int(candle["lowPrice"]),
"close": int(candle["closePrice"]),
"volume": int(candle["volume"]),
"value": pd.NA,
"cum_volume": 0,
"cum_value": pd.NA,
}
)
# 누적 거래량 계산 (과거 → 최신)
cum_volume = 0
for row in reversed(rows):
cum_volume += row["volume"]
row["cum_volume"] = cum_volume
return _create_minute_dataframe(rows)
[docs]
def get_orderbook(self, asset_code: str, data_exchange: DataExchange = DataExchange.KRX) -> Dict:
"""
특정 종목의 호가/잔량 정보를 조회하여 반환합니다.
Note:
특정 시간대(단일가 구간 등)에는 10호가 미만으로 응답될 수 있으며, 이 경우
부족한 항목을 ``{"price": 0, "volume": 0}`` 으로 채워 항상 10호가를 반환합니다.
(KIS 와 동일한 형태)
Args:
asset_code (str): 조회할 자산의 종목번호.
data_exchange (DataExchange): 미지원. 통합(KRX+NXT) 시세만 제공됨
Returns:
dict: 호가 정보가 포함된 사전.
- total_bid_volume (int): 총 매수 잔량.
- total_ask_volume (int): 총 매도 잔량.
- ask_price (int): 1차 매도 호가 가격.
- ask_volume (int): 1차 매도 호가 잔량.
- bid_price (int): 1차 매수 호가 가격.
- bid_volume (int): 1차 매수 호가 잔량.
- time (dtm.datetime): 호가 정보 조회 시간.
- bids (list): 매수 호가 목록 (각 항목은 price와 volume을 포함하는 dict).
- asks (list): 매도 호가 목록 (각 항목은 price과 volume을 포함하는 dict).
"""
r = self.stock_api.get_orderbook(asset_code)
if not r or (not r.get("asks") and not r.get("bids")):
return {}
asks = [{"price": int(entry["price"]), "volume": int(entry["volume"])} for entry in r.get("asks") or []]
bids = [{"price": int(entry["price"]), "volume": int(entry["volume"])} for entry in r.get("bids") or []]
# 10호가 미만으로 응답되는 시간대에는 빈 호가를 0 으로 채운다
asks.extend({"price": 0, "volume": 0} for _ in range(_ORDERBOOK_DEPTH - len(asks)))
bids.extend({"price": 0, "volume": 0} for _ in range(_ORDERBOOK_DEPTH - len(bids)))
return {
"total_bid_volume": sum(entry["volume"] for entry in bids),
"total_ask_volume": sum(entry["volume"] for entry in asks),
"ask_price": asks[0]["price"] if asks else 0,
"ask_volume": asks[0]["volume"] if asks else 0,
"bid_price": bids[0]["price"] if bids else 0,
"bid_volume": bids[0]["volume"] if bids else 0,
"time": r.get("timestamp"),
"bids": bids,
"asks": asks,
}
# ------------------------------------------------------------------ #
# 주문
# ------------------------------------------------------------------ #
@staticmethod
def _order_side_str(side: OrderSide) -> str:
if side == OrderSide.BUY:
return "BUY"
elif side == OrderSide.SELL:
return "SELL"
else:
raise ValueError("지원하지 않는 주문 방향입니다.")
@staticmethod
def _order_side(side: str) -> OrderSide:
if side == "BUY":
return OrderSide.BUY
elif side == "SELL":
return OrderSide.SELL
else:
raise ValueError(f"지원하지 않는 주문 방향입니다. {side}")
@staticmethod
def _order_type_str(order_type: OrderType) -> str:
if order_type == OrderType.LIMIT:
return "LIMIT"
elif order_type == OrderType.MARKET:
return "MARKET"
else:
raise ValueError("지원하지 않는 주문 유형입니다.")
@staticmethod
def _order_type(order_type: str) -> OrderType:
if order_type == "LIMIT":
return OrderType.LIMIT
elif order_type == "MARKET":
return OrderType.MARKET
else:
raise ValueError(f"지원하지 않는 주문 유형입니다. {order_type}")
[docs]
def create_order(
self,
asset_code: str,
side: OrderSide,
quantity: int,
order_type: OrderType = OrderType.MARKET,
price: int = 0,
stop_price: int = 0,
exchange: OrderExchange = OrderExchange.KRX,
confirm_high_value_order: bool = True,
) -> str:
"""
주문을 생성합니다.
Args:
asset_code (str): 종목 코드
side (OrderSide): 주문 방향
quantity (int): 주문 수량
order_type (OrderType): 주문 유형 (LIMIT/MARKET 만 지원)
price (int): 주문 가격 (지정가 주문일 경우에만 필요)
stop_price (int): 미지원 - 0 이외의 값이면 ValueError
exchange (OrderExchange): 미지원 (통합거래소로 주문됨)
confirm_high_value_order (bool): 고액(1억원 이상) 주문 확인 플래그
Returns:
str: 주문 번호
"""
if stop_price != 0:
raise ValueError("스톱 주문은 지원되지 않습니다.")
side_str = self._order_side_str(side)
order_type_str = self._order_type_str(order_type)
if order_type == OrderType.LIMIT:
if int(price) <= 0:
raise ValueError("지정가 주문은 가격이 필요합니다.")
order_price = int(price)
else:
order_price = None
r = self.stock_api.create_order(self.account_no, asset_code, side_str, order_type_str, quantity, price=order_price, confirm_high_value_order=confirm_high_value_order)
return r["orderId"]
[docs]
def update_order(
self,
org_order_no: str,
order_type: OrderType,
price: int,
quantity: int = 0,
stop_price: int = 0,
exchange: OrderExchange = OrderExchange.KRX,
) -> str:
"""
주문을 정정합니다.
Note:
정정 시 새로운 주문 ID가 발급되며, 원주문은 REPLACED 상태가 됩니다.
국내 주식 정정은 수량이 필수라, quantity 미지정(0) 시 원주문의 미체결 수량으로 정정합니다.
Args:
org_order_no (str): 원주문번호
order_type (OrderType): 주문 유형 (LIMIT/MARKET 만 지원)
price (int): 정정 가격
quantity (int): 정정 수량 (0이면 원주문의 미체결 수량 전체)
stop_price (int): 미지원 - 0 이외의 값이면 ValueError
exchange (OrderExchange): 미지원
Returns:
str: 주문 번호 (새로 발급된 주문 ID)
"""
if stop_price != 0:
raise ValueError("스톱 주문은 지원되지 않습니다.")
order_type_str = self._order_type_str(order_type)
if quantity == 0:
order = self.get_order(org_order_no)
if order is None:
raise ValueError(f"주문을 찾을 수 없습니다. {org_order_no}")
if order.pending_quantity <= 0:
raise ValueError(f"정정 가능한 미체결 수량이 없습니다. {org_order_no}")
quantity = order.pending_quantity
if order_type == OrderType.LIMIT:
if int(price) <= 0:
raise ValueError("지정가 주문은 가격이 필요합니다.")
order_price = int(price)
else:
order_price = None
r = self.stock_api.modify_order(self.account_no, org_order_no, order_type_str, quantity=quantity, price=order_price)
return r["orderId"]
[docs]
def cancel_order(self, org_order_no: str, quantity: int = 0, stop_price: int = 0, exchange: OrderExchange = OrderExchange.KRX) -> str:
"""
주문을 취소합니다.
Args:
org_order_no (str): 원주문번호
quantity (int): 미지원 - 전량 취소만 가능하므로 0 이외의 값이면 ValueError
stop_price (int): 미지원
exchange (OrderExchange): 미지원
Returns:
str: 주문 번호
"""
if quantity not in (0, None):
raise ValueError("부분 취소는 지원되지 않습니다.")
if stop_price != 0:
raise ValueError("스톱 주문은 지원되지 않습니다.")
r = self.stock_api.cancel_order(self.account_no, org_order_no)
return r.get("orderId", org_order_no) if isinstance(r, dict) else org_order_no
# ------------------------------------------------------------------ #
# 주문 조회
# ------------------------------------------------------------------ #
def _to_stock_order(self, item: dict, current_price: int = 0) -> StockOrder:
"""주문 응답을 StockOrder 로 변환한다."""
execution = item.get("execution") or {}
quantity = int(item.get("quantity") or 0)
filled_quantity = int(execution.get("filledQuantity") or 0)
status = item.get("status")
is_pending = status in _PENDING_STATUSES
average_filled_price = execution.get("averageFilledPrice")
return StockOrder(
order_no=item["orderId"],
asset_code=item["symbol"],
side=self._order_side(item["side"]),
price=int(item.get("price") or 0), # MARKET 주문은 price 없음 -> 0
quantity=quantity,
filled_quantity=filled_quantity,
pending_quantity=max(0, quantity - filled_quantity) if is_pending else 0,
order_time=item.get("orderedAt"),
filled_price=average_filled_price if average_filled_price is not None else 0,
current_price=current_price,
is_pending=is_pending,
org_order_no=None,
order_type=self._order_type(item["orderType"]),
req_type=OrderRequestType.NEW, # 정정/취소 주문 체인은 노출되지 않음
exchange=None, # 체결 거래소는 노출되지 않음
)
def _fetch_current_prices(self, asset_codes: List[str]) -> Dict[str, int]:
"""종목별 현재가 조회 (200개씩 분할)"""
result = {}
codes = list(dict.fromkeys(asset_codes))
for i in range(0, len(codes), _MAX_SYMBOLS_PER_REQUEST):
for item in self.stock_api.get_prices(codes[i : i + _MAX_SYMBOLS_PER_REQUEST]):
result[item["symbol"]] = int(item["lastPrice"])
return result
[docs]
def get_pending_orders(self, exchanges: List[OrderExchange] = [OrderExchange.KRX]) -> List[StockOrder]:
"""
미체결 주문을 조회합니다.
Args:
exchanges (List[OrderExchange]): 미지원
Returns:
List[StockOrder]: 미체결 주문 리스트
"""
r = self.stock_api.get_orders(self.account_no, {"status": "OPEN"})
orders = [self._to_stock_order(item) for item in r.get("orders") or []]
if orders:
price_map = self._fetch_current_prices([order.asset_code for order in orders])
for order in orders:
order.current_price = price_map.get(order.asset_code, 0)
return orders
[docs]
def get_today_order_history(self, target_date: dtm.date = None, order_no: str = "", exchanges: List[OrderExchange] = [OrderExchange.KRX]) -> List[StockOrder]:
"""
오늘(또는 지정 일자) 주문 내역을 조회합니다.
Args:
target_date (dtm.date): 조회할 날짜 (기본: 오늘)
order_no (str): 조회할 주문 번호. 지정 시 해당 주문만 조회.
exchanges (List[OrderExchange]): 미지원
Returns:
List[StockOrder]: 주문 내역 리스트 (주문 시각 역순)
"""
if order_no:
order = self.get_order(order_no)
return [order] if order is not None else []
if target_date is None:
target_date = dtm.date.today()
date_str = target_date.isoformat()
merged = {}
r = self.stock_api.get_orders(self.account_no, {"status": "OPEN", "from": date_str, "to": date_str})
for item in r.get("orders") or []:
merged[item["orderId"]] = item
# CLOSED 는 커서 페이지네이션. OPEN 조회 이후 체결된 주문이 양쪽에 나타날 수 있어 CLOSED 가 우선한다.
cursor = None
try:
for _ in range(_MAX_PAGES):
params = {"status": "CLOSED", "from": date_str, "to": date_str, "limit": 100}
if cursor is not None:
params["cursor"] = cursor
r = self.stock_api.get_orders(self.account_no, params)
for item in r.get("orders") or []:
merged[item["orderId"]] = item
next_cursor = r.get("nextCursor")
if not r.get("hasNext") or next_cursor is None or next_cursor == cursor:
break
cursor = next_cursor
except requests.HTTPError as e:
# CLOSED 조회가 아직 지원되지 않는 경우(400 closed-not-supported) 진행 중 주문만 반환한다
if _error_code(e) != "closed-not-supported":
raise
self.logger.warning("종료된 주문(CLOSED) 조회가 지원되지 않아 진행 중 주문만 반환합니다.")
orders = [self._to_stock_order(item) for item in merged.values()]
orders.sort(key=lambda order: order.order_time, reverse=True)
return orders
[docs]
def get_order(self, order_no: str) -> Optional[StockOrder]:
"""
주문 번호로 주문 정보를 조회합니다.
Args:
order_no (str): 주문 번호
Returns:
StockOrder: 주문 정보. 찾을 수 없으면 None.
"""
try:
item = self.stock_api.get_order(self.account_no, order_no)
except requests.HTTPError as e:
# 404(order-not-found)만 None 반환, 그 외(권한/계좌 오류 등)는 그대로 전파
if e.response is not None and e.response.status_code == 404:
self.logger.debug(f"get_order({order_no}) 조회 실패: {_error_code(e)}")
return None
raise
return self._to_stock_order(item)
# ------------------------------------------------------------------ #
# 주문 이벤트
# ------------------------------------------------------------------ #
[docs]
async def listen_order_event(self, stop_event: Optional[asyncio.Event] = None) -> AsyncGenerator:
"""
계좌 주문 이벤트를 수신하는 메서드
Raises:
NotImplementedError: 실시간 주문 이벤트(웹소켓)는 아직 지원되지 않습니다.
"""
raise NotImplementedError("실시간 주문 이벤트(웹소켓)는 아직 지원되지 않습니다.")
yield # pragma: no cover - AsyncGenerator 시그니처 유지용