import asyncio
import datetime as dtm
from collections import defaultdict
from dataclasses import asdict, dataclass
from decimal import ROUND_UP, Decimal
from typing import AsyncGenerator, Dict, List, Optional, Union
from zoneinfo import ZoneInfo
import pandas as pd
import requests
from pyqqq.brokerage.toss.oauth import TossAuth
from pyqqq.brokerage.toss.overseas_stock import TossOverseasStock
from pyqqq.brokerage.toss.simple import (
_MAX_PAGES,
_MAX_SYMBOLS_PER_REQUEST,
_ORDERBOOK_DEPTH,
_PENDING_STATUSES,
_error_code,
_order_side,
_order_side_str,
_order_type,
_order_type_str,
)
from pyqqq.datatypes import *
from pyqqq.utils.logger import get_logger
from pyqqq.utils.market_schedule import get_market_schedule
_NYT = ZoneInfo("America/New_York")
_KST = ZoneInfo("Asia/Seoul")
[docs]
@dataclass(kw_only=True)
class TossOverseasStockPosition(OverseasStockPosition):
exchange: Optional[str] = None
""" 거래소 (토스증권은 종목의 거래소를 노출하지 않아 항상 None) """
currency: str = "USD"
""" 통화 """
[docs]
@dataclass(kw_only=True)
class TossOverseasStockOrder(OverseasStockOrder):
exchange: Optional[str] = None
""" 거래소 (토스증권은 체결 거래소를 노출하지 않아 항상 None) """
currency: str = "USD"
""" 통화 """
def _num(value: Optional[Union[int, Decimal]]) -> Union[int, Decimal]:
"""수량 값을 정수이면 int, 소수점이 있으면 Decimal 로 반환한다. None 은 0."""
if value is None:
return 0
d = Decimal(value)
if d == d.to_integral_value():
return int(d)
return d.normalize()
def _to_kr_time(value: Optional[dtm.datetime]) -> Optional[dtm.datetime]:
"""``_coerce`` 가 반환한 KST-naive 시각을 aware KST 시각으로 변환한다."""
return value.replace(tzinfo=_KST) if value is not None else None
def _to_ny_time(value: Optional[dtm.datetime]) -> Optional[dtm.datetime]:
"""``_coerce`` 가 반환한 KST-naive 시각을 aware 미국 동부 시각으로 변환한다."""
return value.replace(tzinfo=_KST).astimezone(_NYT) if value is not None else None
[docs]
class TossSimpleOverseasStock:
"""
토스증권 해외(미국)주식 API를 사용하여 주식 거래를 하기 위한 클래스입니다.
기존 TossOverseasStock 클래스를 감싸고, 간단한 주문/조회 기능을 제공합니다.
메서드 시그니처는 :class:`KISSimpleOverseasStock` 을 따릅니다.
토스증권 Open API 제약으로 인한 차이점:
- 종목의 거래소(NYSE/NASDAQ/AMEX)가 노출되지 않아 포지션/주문의 ``exchange`` 는 항상 ``None``
이며, 심볼은 거래소 접미사 없는 티커("AAPL")를 사용합니다.
- 소수점(0.5주 등) 수량 주문은 시장가 매도(MARKET+SELL)에만 허용되며, 그 외 주문은
정수 수량만 가능합니다. (소수점 매수는 금액(orderAmount) 주문 전용이라 지원되지 않음)
조회 결과의 수량은 정수이면 int, 소수점이 있으면 Decimal 로 반환됩니다.
- 주문 유형은 LIMIT/MARKET 만 지원됩니다. (MOO/LOO/MOC/LOC 미지원)
- 거래대금(value) 데이터가 제공되지 않아 관련 값은 ``None``/``NA`` 로 반환됩니다.
- 예약 주문(``schedule_order`` 계열)과 주간거래(``daytime_order`` 계열)는 지원되지 않아
``NotImplementedError`` 를 발생시킵니다. 실시간 주문 이벤트(``listen_order_event``)도
아직 지원되지 않습니다. (TradingTracker 사용 불가)
- 종목 상세 정보(PER/시가총액 등)가 제공되지 않아 ``get_price_detail`` 은 없습니다.
- 미국 주식 정정은 가격 변경만 지원됩니다. (수량/유형 변경 불가) 정정 시 새로운 주문 ID가
발급되고 원주문은 ``REPLACED`` 상태가 됩니다.
- 취소는 전량 취소만 가능합니다.
- 취소/정정 요청이 별도의 주문 레코드로 조회되지 않아(KIS 와 다름) 주문 내역의
``req_type`` 은 항상 ``NEW``, ``org_order_no`` 는 항상 ``None`` 입니다.
- "오늘"(``get_today_minute_data``, ``get_today_order_history``)은 미국 동부 현지 날짜
기준입니다. (KST 새벽 장중에는 KST 기준 전일이 "오늘")
Args:
auth (TossAuth): 인증 정보
account_no (str): 계좌번호. 미지정 시 첫 번째 계좌를 사용한다. (하나의 App Key로 여러 계좌 지원)
"""
nyt = _NYT
kst = _KST
[docs]
def __init__(self, auth: TossAuth, account_no: Optional[str] = None):
self.auth = auth
self.stock_api = TossOverseasStock(auth)
self._account_no = account_no
self.account_product_code = "" # tracker 호환용
self.currency_code = "USD" # KISSimpleOverseasStock 호환용
self.logger = get_logger(__name__ + ".TossSimpleOverseasStock")
@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_supported_exchange_codes(self) -> List[str]:
"""지원하는 미국 거래소 코드 목록 (참고용 - 토스증권은 거래소를 구분하지 않음)"""
return ["NYSE", "NASD", "AMEX"]
# ------------------------------------------------------------------ #
# 계좌 / 자산
# ------------------------------------------------------------------ #
[docs]
def get_account(self) -> Dict:
"""
계좌 요약 정보를 조회하여 총 잔고, 투자 가능 현금, 매입 금액 및 손익 정보를 반환합니다.
미국 주식 보유분(USD 버킷)만 집계합니다.
Returns:
dict: 계좌 요약 정보
- total_balance (Decimal): 평가 자산 및 미체결 매수 주문을 포함한 총 잔고 (USD)
- investable_cash (Decimal): 신규 주문에 사용 가능한 현금 (USD)
- purchase_amount (Decimal): 현재 보유한 포지션의 총 매입 금액
- evaluated_amount (Decimal): 현재 보유한 포지션의 평가 금액
- pnl_amount (Decimal): 평가 손익
- pnl_rate (Decimal): 손익률 (손익 / 매입 금액 * 100)
"""
holdings = self.stock_api.get_holdings(self.account_no)
buying_power = self.stock_api.get_buying_power(self.account_no, "USD")
purchase_amount = holdings["totalPurchaseAmount"]["usd"] or Decimal(0)
evaluated_amount = holdings["marketValue"]["amount"]["usd"] or Decimal(0)
pnl_amount = (holdings["profitLoss"]["amount"] or {}).get("usd") or Decimal(0)
pnl_rate = pnl_amount / purchase_amount * 100 if purchase_amount != 0 else Decimal(0)
investable_cash = buying_power["cashBuyingPower"] or Decimal(0)
# 미체결 매수 주문에 잡혀 있는 금액 (KIS 해외와 동일하게 총 잔고에 포함)
holding_balance = sum((order.price * order.pending_quantity for order in self.get_pending_orders() if order.side == OrderSide.BUY), Decimal(0))
return {
"total_balance": investable_cash + evaluated_amount + holding_balance,
"investable_cash": investable_cash,
"purchase_amount": purchase_amount,
"evaluated_amount": evaluated_amount,
"pnl_amount": pnl_amount,
"pnl_rate": pnl_rate,
}
[docs]
def get_possible_quantity(self, ticker: str, price: Optional[Decimal] = None) -> Dict:
"""
특정 티커에 대해 지정된 가격으로 주문 가능한 최대 수량과 금액을 조회합니다.
수량은 매매 수수료(센트 단위 올림)를 포함한 총비용이 매수 가능 현금을 넘지 않는
최대치로 계산합니다. 소수점 수량 매수는 지원되지 않아(금액 주문 전용) 정수 수량만
반환합니다.
Args:
ticker (str): 조회할 자산의 티커(symbol)
price (Decimal): 조회할 가격. 지정하지 않으면 현재 가격을 사용.
Returns:
dict: 주문 가능한 수량 및 금액 정보
- currency (str): 거래 통화 코드 ("USD")
- possible_amount (Decimal): 주문 가능한 금액
- quantity (int): 주문 가능한 최대 정수 수량 (수수료 반영)
- price (Decimal): 계산 기준 단가
Raises:
ValueError: 현재가 조회 실패 또는 가격이 유효하지 않은 경우
"""
buying_power = self.stock_api.get_buying_power(self.account_no, "USD")
possible_amount = buying_power["cashBuyingPower"] or Decimal(0)
if price is None:
prices = self.stock_api.get_prices([ticker])
if not prices:
raise ValueError(f"현재가를 조회할 수 없습니다. {ticker}")
price = prices[0]["lastPrice"]
price = Decimal(price)
if price <= 0:
raise ValueError("가격은 0보다 커야 합니다.")
# 미국 수수료율은 수수료 조회 API 값이 정확해 그대로 사용한다 (국내는 부정확 - simple.py FIXME 참조)
commission_rate = self.stock_api._commission_rate(self.account_no)
def total_cost(quantity: Decimal) -> Decimal:
amount = quantity * price
return amount + (amount * commission_rate).quantize(Decimal("0.01"), rounding=ROUND_UP)
quantity = int(possible_amount / (price * (1 + commission_rate)))
while quantity > 0 and total_cost(Decimal(quantity)) > possible_amount:
quantity -= 1
return {
"currency": "USD",
"possible_amount": possible_amount,
"quantity": quantity,
"price": price,
}
[docs]
def get_positions(self, to_frame: bool = False) -> Union[List[TossOverseasStockPosition], pd.DataFrame]:
"""
보유 종목을 조회합니다. (미국 주식만)
Note:
보유 주식 응답에는 매도 가능 수량이 없어, 보유 수량에서 미체결 매도 주문
수량을 차감하여 계산합니다. 담보/대여 등으로 잠긴 수량은 반영되지 않으므로
실제보다 클 수 있습니다.
Args:
to_frame (bool): True 일 경우 asset_code 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[TossOverseasStockPosition] | pd.DataFrame: 보유 종목 정보
"""
holdings = self.stock_api.get_holdings(self.account_no)
items = [item for item in (holdings.get("items") or []) if item["quantity"] > 0]
pending_sell = defaultdict(lambda: Decimal(0))
if items:
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 = order.get("quantity") or Decimal(0)
filled = (order.get("execution") or {}).get("filledQuantity") or Decimal(0)
pending_sell[order["symbol"]] += max(Decimal(0), quantity - filled)
positions = [self._to_stock_position(item, pending_sell[item["symbol"]]) for item in items]
if to_frame:
columns = ["asset_code", "asset_name", "quantity", "sell_possible_quantity", "average_purchase_price", "current_price", "current_value", "current_pnl", "current_pnl_value", "exchange", "currency"]
rows = []
for position in positions:
d = asdict(position)
for key in ("quantity", "sell_possible_quantity", "average_purchase_price", "current_price", "current_value", "current_pnl", "current_pnl_value"):
d[key] = float(d[key]) if d[key] is not None else None
rows.append(d)
return pd.DataFrame(rows, columns=columns).set_index("asset_code")
return positions
@staticmethod
def _to_stock_position(item: dict, pending_sell_quantity: Decimal = Decimal(0)) -> TossOverseasStockPosition:
"""보유 종목 응답 항목을 TossOverseasStockPosition 으로 변환한다."""
quantity = item["quantity"] or Decimal(0)
return TossOverseasStockPosition(
asset_code=item["symbol"],
asset_name=item.get("name", ""),
quantity=_num(quantity),
sell_possible_quantity=_num(max(Decimal(0), quantity - pending_sell_quantity)),
average_purchase_price=item["averagePurchasePrice"],
current_price=item["lastPrice"],
current_value=item["marketValue"]["amount"],
current_pnl=Decimal(item["profitLoss"]["rate"]) * 100, # 소수비율(0.1077=10.77%)로 내려와 % 단위로 변환
current_pnl_value=item["profitLoss"]["amount"],
currency=item.get("currency") or "USD",
)
# ------------------------------------------------------------------ #
# 시세
# ------------------------------------------------------------------ #
@staticmethod
def _candle_time(candle: dict) -> dtm.datetime:
"""캔들 timestamp(KST-naive)를 미국 동부 aware 시각으로 변환한다.
timestamp 는 실제 시각의 KST 표현이다. (일봉은 미국 동부 자정 기준으로 내려와
NY 변환 날짜가 실제 미국 거래일과 일치함 - 2026-07-23 라이브 검증)
"""
return candle["timestamp"].replace(tzinfo=_KST).astimezone(_NYT)
def _iter_candles(self, ticker: 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(ticker, interval=interval, count=200, before=before, adjusted=adjusted)
candles = page.get("candles") or []
if not candles:
return
for candle in candles:
if self._candle_time(candle).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, ticker: str, first_date: dtm.date, last_date: dtm.date, period: str = "D", adjusted_price: bool = True) -> pd.DataFrame:
"""
일봉 데이터 검색
Note:
거래대금이 제공되지 않아 ``value`` 컬럼은 ``NA`` 입니다.
Args:
ticker (str): 티커(symbol)
first_date (datetime.date): 조회 시작일자 (미국 현지 기준)
last_date (datetime.date): 조회 종료일자 (미국 현지 기준)
period (str): 조회 주기. "D"(일간)만 지원.
adjusted_price (bool): 수정 주가 여부
Returns:
pd.DataFrame: 일봉 데이터 (index: date, 시간 순)
"""
if period != "D":
raise ValueError(f"지원하지 않는 주기입니다. {period}")
assert first_date <= last_date, "last_date는 first_date와 같거나, 이후 날짜여야 합니다"
rows = []
for candle in self._iter_candles(ticker, "1d", first_date, adjusted=adjusted_price):
date = self._candle_time(candle).date()
if date > last_date:
continue
rows.append(
{
"date": date,
"open": float(candle["openPrice"]),
"high": float(candle["highPrice"]),
"low": float(candle["lowPrice"]),
"close": float(candle["closePrice"]),
"volume": int(candle["volume"]),
"value": pd.NA,
}
)
rows.reverse() # 과거 -> 최신 (KIS 해외와 동일)
return pd.DataFrame(rows, columns=["date", "open", "high", "low", "close", "volume", "value"]).set_index("date")
[docs]
def get_today_minute_data(self, ticker: str) -> pd.DataFrame:
"""
분봉 데이터 검색 (미국 현지 기준 오늘)
Note:
거래대금이 제공되지 않아 ``value`` 컬럼은 ``NA`` 입니다.
휴장일에는 빈 DataFrame을 반환합니다.
Args:
ticker (str): 티커(symbol)
Returns:
pd.DataFrame: 분봉 데이터 (index: time(미국 동부), kr_time(KST) 포함, 시간 순)
"""
def _create_minute_dataframe(data: Optional[list] = None) -> pd.DataFrame:
df = pd.DataFrame(data or [], columns=["time", "kr_time", "open", "high", "low", "close", "volume", "value"])
df.set_index("time", inplace=True)
return df
today = dtm.datetime.now(_NYT).date()
schedule = get_market_schedule(today, exchange="NYSE")
if schedule.full_day_closed:
return _create_minute_dataframe()
rows = []
for candle in self._iter_candles(ticker, "1m", today):
rows.append(
{
"time": self._candle_time(candle).replace(tzinfo=None),
"kr_time": candle["timestamp"],
"open": float(candle["openPrice"]),
"high": float(candle["highPrice"]),
"low": float(candle["lowPrice"]),
"close": float(candle["closePrice"]),
"volume": int(candle["volume"]),
"value": pd.NA,
}
)
rows.reverse() # 과거 -> 최신 (KIS 해외와 동일)
return _create_minute_dataframe(rows)
[docs]
def get_price(self, ticker: str) -> pd.DataFrame:
"""
특정 티커의 현재 가격 정보를 조회하여 데이터프레임으로 반환합니다.
Note:
거래대금이 제공되지 않아 ``cum_value`` 는 ``NA``, 매수 주문 가능 여부가
제공되지 않아 ``ordy`` 는 ``None`` 입니다. 당일 캔들이 아직 없으면
누적 거래량은 0 입니다.
Args:
ticker (str): 티커(symbol)
Returns:
pd.DataFrame: 현재 가격 정보 (index: ticker)
- current_price (float): 현재 가격
- cum_volume (int): 누적 거래량
- cum_value (NA): 누적 거래 금액 (제공되지 않음)
- diff (float): 전일 종가 대비 가격 차이
- diff_rate (float): 전일 종가 대비 등락률
- sign (int): 등락 기호 (2: 상승, 3: 보합, 5: 하락. 미국 시장은 상/하한 없음)
- pclose (float): 전일 종가
- pvolume (int): 전일 거래량
- ordy (None): 매수 주문 가능 여부 (제공되지 않음)
Raises:
ValueError: 티커를 찾을 수 없는 경우
"""
prices = self.stock_api.get_prices([ticker])
if not prices:
raise ValueError(f"Ticker {ticker} not found")
current_price = prices[0]["lastPrice"]
candle_page = self.stock_api.get_candles(ticker, interval="1d", count=2)
candles = candle_page.get("candles") or []
today = dtm.datetime.now(_NYT).date()
today_candle = candles[0] if candles and self._candle_time(candles[0]).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
pclose = prev_candle["closePrice"] if prev_candle else None
diff = current_price - pclose if pclose else Decimal(0)
diff_rate = float(diff / pclose * 100) if pclose else 0.0
return pd.DataFrame(
[
{
"ticker": ticker,
"current_price": float(current_price),
"cum_volume": int(today_candle["volume"]) if today_candle else 0,
"cum_value": pd.NA,
"diff": float(diff),
"diff_rate": diff_rate,
"sign": 2 if diff > 0 else 5 if diff < 0 else 3,
"pclose": float(pclose) if pclose is not None else 0.0,
"pvolume": int(prev_candle["volume"]) if prev_candle else 0,
"ordy": None,
}
]
).set_index("ticker")
[docs]
def get_orderbook(self, ticker: str) -> Dict:
"""
특정 종목의 호가/잔량 정보를 조회하여 반환합니다.
Note:
10호가 미만으로 응답되는 경우 부족한 항목을 ``{"price": 0, "volume": 0}`` 으로
채워 항상 10호가를 반환합니다. (KIS 와 동일한 형태)
Args:
ticker (str): 티커(symbol)
Returns:
dict: 호가 정보가 포함된 사전.
- total_bid_volume (int|Decimal): 총 매수 잔량.
- total_ask_volume (int|Decimal): 총 매도 잔량.
- ask_price (Decimal): 1차 매도 호가 가격.
- ask_volume (int|Decimal): 1차 매도 호가 잔량.
- bid_price (Decimal): 1차 매수 호가 가격.
- bid_volume (int|Decimal): 1차 매수 호가 잔량.
- time (dtm.datetime): 미국 동부 기준 호가 정보 조회 시간.
- bids (list): 매수 호가 목록 (각 항목은 price와 volume을 포함하는 dict).
- asks (list): 매도 호가 목록 (각 항목은 price과 volume을 포함하는 dict).
"""
r = self.stock_api.get_orderbook(ticker)
if not r or (not r.get("asks") and not r.get("bids")):
return {}
asks = [{"price": entry["price"], "volume": _num(entry["volume"])} for entry in r.get("asks") or []]
bids = [{"price": entry["price"], "volume": _num(entry["volume"])} for entry in r.get("bids") or []]
# 10호가 미만으로 응답되는 경우에는 빈 호가를 0 으로 채운다
asks.extend({"price": Decimal(0), "volume": 0} for _ in range(_ORDERBOOK_DEPTH - len(asks)))
bids.extend({"price": Decimal(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"],
"ask_volume": asks[0]["volume"],
"bid_price": bids[0]["price"],
"bid_volume": bids[0]["volume"],
"time": _to_ny_time(r.get("timestamp")),
"bids": bids,
"asks": asks,
}
# ------------------------------------------------------------------ #
# 주문
# ------------------------------------------------------------------ #
[docs]
def create_order(self, ticker: str, side: OrderSide, quantity: Union[int, Decimal], order_type: OrderType, price: Decimal = Decimal("0"), confirm_high_value_order: bool = True) -> str:
"""
주문을 생성합니다.
Args:
ticker (str): 티커(symbol)
side (OrderSide): 주문 방향
quantity (int|Decimal): 주문 수량. 소수점 수량은 시장가 매도(MARKET+SELL)에만 허용.
order_type (OrderType): 주문 유형 (LIMIT/MARKET 만 지원)
price (Decimal): 주문 가격 (지정가 주문일 경우에만 필요)
confirm_high_value_order (bool): 고액(1억원 상당 이상) 주문 확인 플래그
Returns:
str: 주문 번호
Raises:
ValueError: 지원하지 않는 주문 유형/수량/가격인 경우
"""
side_str = _order_side_str(side)
order_type_str = _order_type_str(order_type) # LIMIT/MARKET 외에는 ValueError
quantity = Decimal(quantity)
if quantity <= 0:
raise ValueError("주문 수량은 0보다 커야 합니다.")
if quantity != quantity.to_integral_value() and not (order_type == OrderType.MARKET and side == OrderSide.SELL):
raise ValueError("소수점 수량 주문은 시장가 매도(MARKET+SELL)에만 사용할 수 있습니다.")
if order_type == OrderType.LIMIT:
if Decimal(price) <= 0:
raise ValueError("지정가 주문은 가격이 필요합니다.")
order_price = price
else:
order_price = None
r = self.stock_api.create_order(self.account_no, ticker, 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, ticker: str, org_order_no: str, price: Decimal, quantity: Union[int, Decimal] = 0) -> str:
"""
주문 가격을 정정합니다.
Note:
미국 주식 정정은 가격 변경만 지원되며, 새 가격은 미체결 수량 전체에 적용됩니다.
``ticker`` 와 ``quantity`` 는 KIS 호환을 위해 받지만 무시됩니다.
정정 시 새로운 주문 ID가 발급되며, 원주문은 REPLACED 상태가 됩니다.
새 주문 레코드에는 원주문 정보(org_order_no)가 제공되지 않아 조회만으로는
정정 체인을 복원할 수 없습니다. 추적이 필요하면 반환된 새 주문 ID를 호출 측에서
직접 보관하세요.
Args:
ticker (str): 티커(symbol) - 미사용
org_order_no (str): 원주문번호
price (Decimal): 정정 가격
quantity (int|Decimal): 미사용 (수량 변경은 지원되지 않음 - 항상 미체결 수량 전체)
Returns:
str: 주문 번호 (새로 발급된 주문 ID)
"""
if Decimal(price) <= 0:
raise ValueError("정정 가격이 필요합니다.")
r = self.stock_api.modify_order(self.account_no, org_order_no, "LIMIT", price=price)
return r["orderId"]
[docs]
def cancel_order(self, ticker: str, order_no: str, quantity: Union[int, Decimal] = 0) -> str:
"""
주문을 취소합니다.
Note:
토스증권은 전량 취소만 가능하여 ``ticker`` 와 ``quantity`` 는 KIS 호환을 위해
받지만 무시됩니다.
취소는 원주문의 상태를 CANCELED 로 바꿀 뿐, KIS 처럼 별도의 취소 주문 레코드가
주문 내역에 남지 않습니다. 응답으로 원주문과 다른 주문 ID(취소 요청 ID)가
반환되지만 이 ID는 주문 내역 조회에 나타나지 않습니다.
Args:
ticker (str): 티커(symbol) - 미사용
order_no (str): 취소할 주문 번호
quantity (int|Decimal): 미사용 (항상 전량 취소)
Returns:
str: 주문 번호
"""
r = self.stock_api.cancel_order(self.account_no, order_no)
return r.get("orderId", order_no) if isinstance(r, dict) else order_no
# ------------------------------------------------------------------ #
# 주문 조회
# ------------------------------------------------------------------ #
def _to_stock_order(self, item: dict, current_price: Optional[Decimal] = None) -> TossOverseasStockOrder:
"""주문 응답을 TossOverseasStockOrder 로 변환한다."""
execution = item.get("execution") or {}
quantity = item.get("quantity") or Decimal(0)
filled_quantity = execution.get("filledQuantity") or Decimal(0)
status = item.get("status")
is_pending = status in _PENDING_STATUSES
average_filled_price = execution.get("averageFilledPrice")
order_kr_time = _to_kr_time(item.get("orderedAt"))
return TossOverseasStockOrder(
order_no=item["orderId"],
asset_code=item["symbol"],
side=_order_side(item["side"]),
price=item.get("price") or Decimal(0), # MARKET 주문은 price 없음 -> 0
quantity=_num(quantity),
filled_quantity=_num(filled_quantity),
pending_quantity=_num(max(Decimal(0), quantity - filled_quantity)) if is_pending else 0,
order_time=order_kr_time.astimezone(_NYT) if order_kr_time else None,
order_kr_time=order_kr_time,
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=_order_type(item["orderType"]),
req_type=OrderRequestType.NEW, # 정정/취소 주문 체인은 노출되지 않음
currency=item.get("currency") or "USD",
)
def _fetch_current_prices(self, asset_codes: List[str]) -> Dict[str, Decimal]:
"""종목별 현재가 조회 (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"]] = item["lastPrice"]
return result
def _to_orders(self, items: List[dict]) -> List[TossOverseasStockOrder]:
"""주문 응답 목록을 현재가를 보강한 TossOverseasStockOrder 목록으로 변환한다."""
price_map = self._fetch_current_prices([item["symbol"] for item in items]) if items else {}
return [self._to_stock_order(item, price_map.get(item["symbol"])) for item in items]
@staticmethod
def _orders_to_frame(orders: List[TossOverseasStockOrder]) -> pd.DataFrame:
"""주문 목록을 order_no 를 인덱스로 한 DataFrame 으로 변환한다."""
columns = ["order_no", "asset_code", "side", "price", "quantity", "filled_quantity", "pending_quantity", "order_time", "filled_price", "current_price", "is_pending", "org_order_no", "order_type", "req_type", "exchange", "currency", "reject_reason", "order_kr_time"]
rows = []
for order in orders:
d = asdict(order)
for key in ("price", "quantity", "filled_quantity", "pending_quantity", "filled_price", "current_price"):
d[key] = float(d[key]) if d[key] is not None else None
d["side"] = "BUY" if d["side"] == OrderSide.BUY else "SELL"
d["order_type"] = _order_type_str(d["order_type"])
d["req_type"] = "NEW"
rows.append(d)
return pd.DataFrame(rows, columns=columns).astype({"org_order_no": "string", "order_no": "string"}).set_index("order_no")
[docs]
def get_pending_orders(self, to_frame: bool = False) -> Union[List[TossOverseasStockOrder], pd.DataFrame]:
"""
미체결 주문을 조회합니다.
Args:
to_frame (bool): True 일 경우 order_no 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[TossOverseasStockOrder] | pd.DataFrame: 미체결 주문 정보
"""
r = self.stock_api.get_orders(self.account_no, {"status": "OPEN"})
orders = self._to_orders(r.get("orders") or [])
return self._orders_to_frame(orders) if to_frame else orders
[docs]
def get_today_order_history(self, target_date: Optional[dtm.date] = None, to_frame: bool = False) -> Union[List[TossOverseasStockOrder], pd.DataFrame]:
"""
미국 현지 기준 오늘(또는 지정 일자)의 주문 내역을 조회합니다.
Note:
KIS 와 달리 취소/정정 요청은 별도 레코드로 포함되지 않습니다. 취소된 주문은
원주문 레코드가 ``is_pending=False`` 로 조회될 뿐이며, 모든 레코드의
``req_type`` 은 ``NEW``, ``org_order_no`` 는 ``None`` 입니다.
Args:
target_date (dtm.date): 조회할 미국 현지 기준 날짜 (기본: 오늘)
to_frame (bool): True 일 경우 order_no 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[TossOverseasStockOrder] | pd.DataFrame: 주문 내역 (주문 시각 역순)
"""
if target_date is None:
target_date = dtm.datetime.now(_NYT).date()
return self.get_order_history(target_date, target_date, to_frame)
[docs]
def get_order_history(self, from_date: dtm.date, to_date: dtm.date, to_frame: bool = False) -> Union[List[TossOverseasStockOrder], pd.DataFrame]:
"""
주문 내역을 조회합니다.
Note:
KIS 와 달리 취소/정정 요청은 별도 레코드로 포함되지 않습니다. 취소된 주문은
원주문 레코드가 ``is_pending=False`` 로 조회될 뿐이며, 모든 레코드의
``req_type`` 은 ``NEW``, ``org_order_no`` 는 ``None`` 입니다.
Args:
from_date (dtm.date): 조회 시작 날짜 (미국 현지 기준)
to_date (dtm.date): 조회 종료 날짜 (미국 현지 기준)
to_frame (bool): True 일 경우 order_no 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[TossOverseasStockOrder] | pd.DataFrame: 주문 내역 (주문 시각 역순)
"""
assert from_date <= to_date, "to_date는 from_date와 같거나, 이후 날짜여야 합니다"
# 토스 from/to 필터는 orderedAt 의 KST 날짜 기준. 미국 현지 날짜 [from, to] 는
# KST 날짜로 [from, to+1] 에 걸치므로 넓혀 조회한 뒤 현지 날짜로 재필터한다.
query = {"from": from_date.isoformat(), "to": (to_date + dtm.timedelta(days=1)).isoformat()}
merged = {}
r = self.stock_api.get_orders(self.account_no, {"status": "OPEN", **query})
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", "limit": 100, **query}
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_orders(list(merged.values()))
orders = [order for order in orders if order.order_time is not None and from_date <= order.order_time.date() <= to_date]
orders.sort(key=lambda order: order.order_time, reverse=True)
return self._orders_to_frame(orders) if to_frame else orders
[docs]
def get_order(self, order_no: str) -> Optional[TossOverseasStockOrder]:
"""
주문 번호로 주문 정보를 조회합니다.
Args:
order_no (str): 주문 번호
Returns:
TossOverseasStockOrder: 주문 정보. 찾을 수 없으면 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)
# ------------------------------------------------------------------ #
# 미지원 기능 (KIS 시그니처 호환용)
# ------------------------------------------------------------------ #
def schedule_order(self, ticker: str, side: OrderSide, quantity: Union[int, Decimal], order_type: OrderType, price: Decimal) -> str:
"""예약 주문 - 토스증권 Open API 에서 지원되지 않습니다."""
raise NotImplementedError("토스증권은 예약 주문을 지원하지 않습니다.")
def cancel_scheduled_order(self, org_order_no: str, reservation_date: dtm.date):
"""예약 주문 취소 - 토스증권 Open API 에서 지원되지 않습니다."""
raise NotImplementedError("토스증권은 예약 주문을 지원하지 않습니다.")
def get_scheduled_orders(self, start_date: Optional[dtm.date] = None, end_date: Optional[dtm.date] = None, include_cancelled: bool = False, to_frame: bool = False):
"""예약 주문 조회 - 토스증권 Open API 에서 지원되지 않습니다."""
raise NotImplementedError("토스증권은 예약 주문을 지원하지 않습니다.")
def daytime_order(self, ticker: str, side: OrderSide, quantity: Union[int, Decimal], order_type: OrderType, price: Decimal) -> str:
"""주간거래 주문 - 토스증권 Open API 에서 지원되지 않습니다."""
raise NotImplementedError("토스증권은 주간거래(별도 세션) 주문을 지원하지 않습니다.")
def daytime_update_order(self, ticker: str, org_order_no: str, price: Decimal, quantity: Union[int, Decimal]) -> str:
"""주간거래 주문 정정 - 토스증권 Open API 에서 지원되지 않습니다."""
raise NotImplementedError("토스증권은 주간거래(별도 세션) 주문을 지원하지 않습니다.")
def daytime_cancel_order(self, ticker: str, order_no: str, quantity: Union[int, Decimal]) -> str:
"""주간거래 주문 취소 - 토스증권 Open API 에서 지원되지 않습니다."""
raise NotImplementedError("토스증권은 주간거래(별도 세션) 주문을 지원하지 않습니다.")
async def listen_order_event(self, stop_event: Optional[asyncio.Event] = None) -> AsyncGenerator:
"""
계좌 주문 이벤트를 수신하는 메서드
Raises:
NotImplementedError: 실시간 주문 이벤트(웹소켓)는 아직 지원되지 않습니다.
"""
raise NotImplementedError("실시간 주문 이벤트(웹소켓)는 아직 지원되지 않습니다.")
yield # pragma: no cover - AsyncGenerator 시그니처 유지용