from pyqqq.brokerage.toss.oauth import TossAuth
from pyqqq.brokerage.toss.tr_client import TossTRClient
from pyqqq.utils.limiter import CallLimiter
from decimal import Decimal
from typing import List, Optional, Union
import re
import datetime
import pytz
_KST = pytz.timezone("Asia/Seoul")
# API 그룹별 초당 최대 요청 수 (클라이언트 × 그룹 단위 TPS 제한).
# https://developers.tossinvest.com/docs 의 Rate Limits 참조.
_RATE_LIMITS = {
"MARKET_DATA": 10,
"MARKET_DATA_CHART": 5,
"STOCK": 5,
"MARKET_INFO": 3,
"ACCOUNT": 1,
"ASSET": 5,
"ORDER": 6,
"ORDER_HISTORY": 5,
"ORDER_INFO": 6,
}
# 개장 직후(09:00~09:10 KST) 한도가 축소되는 그룹
_PEAK_RATE_LIMITS = {"ORDER": 3, "ORDER_INFO": 3}
_PEAK_START = datetime.time(9, 0)
_PEAK_END = datetime.time(9, 10)
# 응답 필드의 형 변환 대상. 숫자/시각/날짜는 모두 문자열로 내려오며,
# 필드명이 엔드포인트 전반에서 일관되므로 이름 기준으로 변환한다. (세 집합 간 이름 충돌 없음)
_DECIMAL_FIELDS = frozenset(
[
"price",
"volume",
"lastPrice",
"openPrice",
"highPrice",
"lowPrice",
"closePrice",
"upperLimitPrice",
"lowerLimitPrice",
"sharesOutstanding",
"leverageFactor",
"rate",
"rateAfterCost",
"midRate",
"basisPoint",
"commissionRate",
"quantity",
"averagePurchasePrice",
"purchaseAmount",
"amount",
"amountAfterCost",
"krw",
"usd",
"commission",
"tax",
"cashBuyingPower",
"sellableQuantity",
"filledQuantity",
"averageFilledPrice",
"filledAmount",
"orderAmount",
]
)
_DATETIME_FIELDS = frozenset(["timestamp", "validFrom", "validUntil", "startTime", "endTime", "singlePriceAuctionStartTime", "singlePriceAuctionEndTime", "orderedAt", "canceledAt", "filledAt"])
_DATE_FIELDS = frozenset(["date", "listDate", "delistDate", "startDate", "endDate", "settlementDate"])
def _digits(s: str) -> str:
"""문자열에서 숫자만 추출한다. (계좌번호 비교용 정규화)"""
return re.sub(r"\D", "", s or "")
[docs]
class TossDomesticStock:
"""
토스증권 국내주식 API
4xx 에러는 재시도 없이 ``requests.HTTPError`` 로 즉시 발생하며,
``e.response.json()["error"]["code"]`` 로 에러를 구분할 수 있다.
200 응답에 에러 envelope 이 담긴 경우는 ``ValueError`` 가 발생한다.
"""
[docs]
def __init__(self, auth: TossAuth):
self.auth = auth
self.tr_client = TossTRClient(auth)
self._account_seq_cache = {} # _digits(cano) -> accountSeq
def _resolve_account_seq(self, cano: str):
"""계좌번호(cano)로부터 ``accountSeq`` 를 해석하고 캐싱한다."""
if cano is None:
raise ValueError("cano must be set for account-context API calls")
target = _digits(cano)
if target in self._account_seq_cache:
return self._account_seq_cache[target]
for acc in self.get_accounts():
if _digits(acc.get("accountNo", "")) == target:
self._account_seq_cache[target] = acc["accountSeq"]
return acc["accountSeq"]
raise ValueError(f"Account not found for cano={cano}")
@staticmethod
def _limit_rate(group: str, now: Optional[datetime.time] = None) -> int:
"""그룹의 현재 시각 기준 초당 최대 요청 수를 반환한다."""
if group in _PEAK_RATE_LIMITS:
if now is None:
now = datetime.datetime.now(_KST).time()
if _PEAK_START <= now < _PEAK_END:
return _PEAK_RATE_LIMITS[group]
return _RATE_LIMITS[group]
def _wait_limit_rate(self, group: str):
"""API 그룹별 rate limit 을 준수하도록 필요 시 대기한다."""
CallLimiter().wait_limit_rate(self._limit_rate(group), scope=f"toss/{group}")
def _tr_request(self, path: str, method: str = "GET", params: dict = None, body: dict = None, cano: Optional[str] = None, limit_group: Optional[str] = None):
account_seq = self._resolve_account_seq(cano) if cano is not None else None
if limit_group is not None:
self._wait_limit_rate(limit_group)
res = self.tr_client.request(path, method=method, params=params, body=body, account_seq=account_seq)
return self._coerce(self._result(res))
@staticmethod
def _result(res: dict):
"""응답 envelope 를 검사하고 ``result`` 값을 반환한다."""
if isinstance(res, dict) and "error" in res:
err = res["error"] or {}
raise ValueError(f"Error: ({err.get('code')}) {err.get('message')}")
if isinstance(res, dict) and "result" in res:
return res["result"]
return res
@classmethod
def _coerce(cls, obj):
"""응답 값을 타입에 맞게 변환한다. (가격→Decimal, 시각→datetime, 날짜→date)
중첩된 dict/list 를 재귀적으로 순회하며, 알려진 필드명에 한해 문자열 값을
변환한다. 변환 대상이 아니거나 ``None``/빈 문자열/비문자열 값은 그대로 둔다.
"""
if isinstance(obj, list):
return [cls._coerce(x) for x in obj]
if isinstance(obj, dict):
return {k: cls._coerce_field(k, v) for k, v in obj.items()}
return obj
@classmethod
def _coerce_field(cls, key: str, value):
if isinstance(value, (dict, list)):
return cls._coerce(value)
if not isinstance(value, str) or value == "":
return value
if key in _DECIMAL_FIELDS:
return Decimal(value)
if key in _DATETIME_FIELDS:
return datetime.datetime.fromisoformat(value).astimezone(_KST).replace(tzinfo=None)
if key in _DATE_FIELDS:
return datetime.date.fromisoformat(value)
return value
@staticmethod
def _join_symbols(symbols: Union[str, List[str]]) -> str:
"""심볼 목록을 콤마로 구분된 문자열로 변환한다."""
if isinstance(symbols, str):
return symbols
return ",".join(symbols)
# ------------------------------------------------------------------ #
# 시세 (Market Data) - 인증 토큰만 필요, 계좌 헤더 불필요
# ------------------------------------------------------------------ #
[docs]
def get_prices(self, symbols: Union[str, List[str]]):
"""
현재가 조회
Args:
symbols (str|list): 종목코드 또는 종목코드 목록 (최대 200개)
Returns:
list: 종목별 현재가 정보 목록
Raises:
ValueError: API 에러 발생시
"""
params = {"symbols": self._join_symbols(symbols)}
return self._tr_request("/api/v1/prices", params=params, limit_group="MARKET_DATA")
[docs]
def get_candles(self, symbol: str, interval: str = "1d", count: int = 100, before: Optional[str] = None, adjusted: bool = True):
"""
캔들(차트) 데이터 조회
Args:
symbol (str): 종목코드
interval (str): 캔들 주기 - "1m"(분), "1d"(일)
count (int): 조회 개수 (1~200)
before (str): 이 시각(ISO 8601) 이전의 캔들 조회 (페이지네이션)
adjusted (bool): 수정주가 적용 여부
Returns:
dict: 캔들 데이터 (``candles`` 목록과 다음 페이지 커서 ``nextBefore``)
Raises:
ValueError: API 에러 발생시
"""
params = {
"symbol": symbol,
"interval": interval,
"count": count,
"adjusted": str(adjusted).lower(),
}
if before is not None:
params["before"] = before
return self._tr_request("/api/v1/candles", params=params, limit_group="MARKET_DATA_CHART")
[docs]
def get_orderbook(self, symbol: str):
"""
호가 조회
Args:
symbol (str): 종목코드
Returns:
dict: 호가 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/orderbook", params={"symbol": symbol}, limit_group="MARKET_DATA")
[docs]
def get_trades(self, symbol: str):
"""
최근 체결 내역 조회
Args:
symbol (str): 종목코드
Returns:
list: 체결 내역 목록
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/trades", params={"symbol": symbol}, limit_group="MARKET_DATA")
[docs]
def get_price_limits(self, symbol: str):
"""
상/하한가 조회
Args:
symbol (str): 종목코드
Returns:
dict: 상한가/하한가 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/price-limits", params={"symbol": symbol}, limit_group="MARKET_DATA")
[docs]
def get_stocks(self, symbols: Union[str, List[str]]):
"""
종목 기본 정보 조회
Args:
symbols (str|list): 종목코드 또는 종목코드 목록
Returns:
list: 종목 기본 정보 목록
Raises:
ValueError: API 에러 발생시
"""
params = {"symbols": self._join_symbols(symbols)}
return self._tr_request("/api/v1/stocks", params=params, limit_group="STOCK")
[docs]
def get_stock_warnings(self, symbol: str):
"""
종목 투자유의(매수 주의) 정보 조회
Args:
symbol (str): 종목코드
Returns:
dict: 투자유의 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request(f"/api/v1/stocks/{symbol}/warnings", limit_group="STOCK")
[docs]
def get_market_calendar(self, country: str = "KR"):
"""
시장 운영 일정(개장 시간) 조회
Args:
country (str): 국가 코드 - "KR"(국내), "US"(해외)
Returns:
dict: 시장 운영 일정 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request(f"/api/v1/market-calendar/{country}", limit_group="MARKET_INFO")
[docs]
def get_exchange_rate(self):
"""
환율 조회
Returns:
dict: 환율 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/exchange-rate", limit_group="MARKET_INFO")
# ------------------------------------------------------------------ #
# 계좌 / 자산 (Account & Asset) - cano 인자 필요 (get_accounts 제외)
# ------------------------------------------------------------------ #
[docs]
def get_accounts(self):
"""
계좌 목록 조회
반환된 각 계좌의 ``accountNo`` 를 보유주식/주문 등 사용자 컨텍스트 메서드의 ``cano``
인자로 사용한다. (내부적으로 ``accountNo`` → ``accountSeq`` 로 변환되어 헤더에 사용된다.)
Returns:
list: 계좌 정보 목록 (accountNo, accountSeq, accountType)
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/accounts", limit_group="ACCOUNT")
[docs]
def get_holdings(self, cano: str):
"""
보유 주식 조회
Args:
cano (str): 계좌번호
Returns:
dict: 보유 주식 및 평가 정보 (``items`` 는 국내 종목만)
Raises:
ValueError: API 에러 발생시
"""
res = self._tr_request("/api/v1/holdings", cano=cano, limit_group="ASSET")
if isinstance(res, dict) and "items" in res:
res["items"] = [item for item in res["items"] if item.get("marketCountry") == "KR"] # 국내 종목만
return res
[docs]
def get_buying_power(self, cano: str, currency: str = "KRW"):
"""
매수 가능 금액 조회
미수거래를 제외한 현금 기반 매수 가능 금액을 반환한다.
Args:
cano (str): 계좌번호
currency (str): 통화 코드 - "KRW", "USD" (API 필수 파라미터)
Returns:
dict: 매수 가능 금액 정보 (currency, cashBuyingPower)
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/buying-power", params={"currency": currency}, cano=cano, limit_group="ORDER_INFO")
[docs]
def get_sellable_quantity(self, cano: str, symbol: str):
"""
매도 가능 수량 조회
Args:
cano (str): 계좌번호
symbol (str): 종목코드
Returns:
dict: 매도 가능 수량 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/sellable-quantity", params={"symbol": symbol}, cano=cano, limit_group="ORDER_INFO")
# ------------------------------------------------------------------ #
# 주문 (Order) - cano 인자 필요
# ------------------------------------------------------------------ #
[docs]
def create_order(
self,
cano: str,
symbol: str,
side: str,
order_type: str,
quantity: Union[str, int],
price: Optional[Union[str, int]] = None,
time_in_force: Optional[str] = None,
client_order_id: Optional[str] = None,
confirm_high_value_order: Optional[bool] = None,
):
"""
주문 생성
Args:
cano (str): 계좌번호
symbol (str): 종목코드
side (str): 매매 구분 - "BUY"(매수), "SELL"(매도)
order_type (str): 주문 유형 - "LIMIT"(지정가), "MARKET"(시장가)
quantity (str|int): 주문 수량
price (str|int): 주문 가격. 지정가(LIMIT) 주문 시 필수.
time_in_force (str): 주문 조건 - "DAY", "CLS"
client_order_id (str): 사용자 지정 주문 ID (멱등성 키, 10분간 유효)
confirm_high_value_order (bool): 고액 주문 확인 플래그. 1억원 이상 주문 시 True 필수.
Returns:
dict: 생성된 주문 정보 (orderId 등)
Raises:
ValueError: API 에러 발생시
"""
body = {
"symbol": symbol,
"side": side,
"orderType": order_type,
"quantity": str(quantity),
}
if price is not None:
body["price"] = str(price)
if time_in_force is not None:
body["timeInForce"] = time_in_force
if client_order_id is not None:
body["clientOrderId"] = client_order_id
if confirm_high_value_order is not None:
body["confirmHighValueOrder"] = confirm_high_value_order
return self._tr_request("/api/v1/orders", method="POST", body=body, cano=cano, limit_group="ORDER")
[docs]
def modify_order(self, cano: str, order_id: str, order_type: str, quantity: Optional[Union[str, int]] = None, price: Optional[Union[str, int]] = None, confirm_high_value_order: Optional[bool] = None):
"""
주문 정정
Args:
cano (str): 계좌번호
order_id (str): 정정할 주문 ID
order_type (str): 호가 유형 - "LIMIT"(지정가), "MARKET"(시장가) (API 필수 필드)
quantity (str|int): 정정 수량. 국내 주식은 필수 (양의 정수).
price (str|int): 정정 가격. LIMIT 주문 시 필수.
confirm_high_value_order (bool): 고액 주문 확인 플래그. 1억원 이상 주문 시 True 필수.
Returns:
dict: 정정 결과 (orderId)
Raises:
ValueError: API 에러 발생시
"""
body = {"orderType": order_type}
if quantity is not None:
body["quantity"] = str(quantity)
if price is not None:
body["price"] = str(price)
if confirm_high_value_order is not None:
body["confirmHighValueOrder"] = confirm_high_value_order
return self._tr_request(f"/api/v1/orders/{order_id}/modify", method="POST", body=body, cano=cano, limit_group="ORDER")
[docs]
def cancel_order(self, cano: str, order_id: str):
"""
주문 취소
Args:
cano (str): 계좌번호
order_id (str): 취소할 주문 ID
Returns:
dict: 취소 결과
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request(f"/api/v1/orders/{order_id}/cancel", method="POST", cano=cano, limit_group="ORDER")
[docs]
def get_orders(self, cano: str, params: Optional[dict] = None):
"""
주문 내역 조회
API 응답에는 국내/해외 주식 주문이 함께 내려오지만, 해외(USD) 주문은 제외하고
국내(KRW) 주문만 반환한다. 필터로 페이지가 비더라도 ``nextCursor``/``hasNext`` 는
원본 그대로 유지되므로 커서 페이지네이션은 정상 동작한다.
Args:
cano (str): 계좌번호
params (dict): 조회 필터 쿼리 파라미터
- status (str): 필수. "OPEN"(진행 중 주문 전량 반환, limit/cursor 무시) 또는 "CLOSED"(종료 주문, 페이지네이션 적용)
- symbol (str): 종목코드 필터
- from / to (str): 주문 생성일(orderedAt, KST) 범위 필터 (YYYY-MM-DD, inclusive)
- cursor (str): CLOSED 페이지네이션 커서 (응답의 nextCursor)
- limit (int): CLOSED 페이지 크기 (기본 20, 최대 100)
Returns:
dict: 주문 내역 (``orders`` 목록은 국내 주문만, 다음 페이지 커서 ``nextCursor``, ``hasNext``)
Raises:
ValueError: API 에러 발생시
"""
res = self._tr_request("/api/v1/orders", params=params, cano=cano, limit_group="ORDER_HISTORY")
if isinstance(res, dict) and "orders" in res:
res["orders"] = [order for order in res["orders"] if order.get("currency") == "KRW"] # 국내(KRW) 주문만
return res
[docs]
def get_order(self, cano: str, order_id: str):
"""
주문 상세 조회
Args:
cano (str): 계좌번호
order_id (str): 조회할 주문 ID
Returns:
dict: 주문 상세 정보
Raises:
ValueError: 해외 주식 주문을 조회한 경우 또는 API 에러 발생시
"""
res = self._tr_request(f"/api/v1/orders/{order_id}", cano=cano, limit_group="ORDER_HISTORY")
if isinstance(res, dict) and res.get("currency") != "KRW":
raise ValueError(f"해외 주식 주문은 지원되지 않습니다. (orderId={order_id}, currency={res.get('currency')})")
return res
[docs]
def get_commissions(self, cano: str, params: Optional[dict] = None):
"""
거래 수수료 조회
Args:
cano (str): 계좌번호
params (dict): 조회 필터 쿼리 파라미터
Returns:
dict: 수수료 정보
Raises:
ValueError: API 에러 발생시
"""
return self._tr_request("/api/v1/commissions", params=params, cano=cano, limit_group="ORDER_INFO")
def _commission_rate(self, cano: str) -> Decimal:
"""국내 시장에 현재 적용되는 매매 수수료율(소수비율)을 반환한다.
수수료 조회(get_commissions) 결과에서 적용 기간(startDate~endDate, null=무기한)에
오늘이 포함되는 국내(KR) 항목을 찾는다. 적용 항목이 없으면 0.
Args:
cano (str): 계좌번호
Returns:
Decimal: 수수료율 (소수비율. 예: 0.015% -> Decimal("0.00015"))
"""
today = datetime.date.today()
for item in self.get_commissions(cano) or []:
if item.get("marketCountry") != "KR":
continue
start, end = item.get("startDate"), item.get("endDate")
if (start is None or start <= today) and (end is None or today <= end):
return Decimal(item["commissionRate"]) / 100 # API 는 % 단위(0.015 = 0.015%) -> 소수비율로 환산
return Decimal(0)