import asyncio
import datetime as dtm
from collections import defaultdict
from dataclasses import asdict
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")
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")를 사용합니다.
- 소수점 수량 주문은 시장가 매도(MARKET+SELL)에만 허용됩니다. 조회 결과의 수량은 정수이면 int, 소수점이 있으면 Decimal.
- 주문 유형은 LIMIT/MARKET 만 지원됩니다. (MOO/LOO/MOC/LOC 미지원)
- 거래대금(value)은 제공되지 않아 ``None``/``NA`` 로 반환됩니다.
- 예약 주문/주간거래/실시간 주문 이벤트 미지원 - 해당 메서드는 ``NotImplementedError``. (TradingTracker 사용 불가)
- 종목 상세 정보(PER/시가총액 등)가 제공되지 않아 ``get_price_detail`` 은 없습니다.
- 정정은 가격 변경만 지원되고(수량/유형 불가) 새 주문 ID가 발급되며, 취소는 전량 취소만 가능합니다.
- 주문 내역 조회는 취소된 주문에서 취소(CANCEL) 레코드를 분리해 반환합니다.
(``order_no=None``, ``org_order_no`` 로 원주문 연결 - 상세는 문서 "주문 내역과 취소/정정 레코드" 참고)
- "오늘"(``get_today_minute_data``, ``get_today_order_history``)은 미국 동부 현지 날짜 기준입니다.
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[OverseasStockPosition], pd.DataFrame]:
"""
보유 종목을 조회합니다. (미국 주식만)
Note:
매도 가능 수량은 보유 수량에서 미체결 매도 수량을 차감해 계산합니다. (담보/대여 등 잠긴 수량 미반영)
Args:
to_frame (bool): True 일 경우 asset_code 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[OverseasStockPosition] | 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)) -> OverseasStockPosition:
"""보유 종목 응답 항목을 OverseasStockPosition 으로 변환한다."""
quantity = item["quantity"] or Decimal(0)
return OverseasStockPosition(
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"],
exchange=None, # 종목의 거래소는 노출되지 않음
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 상태가 됩니다. 새 주문으로 원주문을
조회할 수 없으니 체인 추적이 필요하면 반환된 주문 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 로 바꾸며, 반환되는 취소 요청 ID로는 조회할 수 없습니다.
취소 내역은 주문 내역 조회가 별도 CANCEL 레코드로 분리해 반환합니다.
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) -> OverseasStockOrder:
"""주문 응답을 OverseasStockOrder 로 변환한다."""
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 OverseasStockOrder(
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, # 취소 레코드는 _to_stock_orders 에서 분리
exchange=None, # 체결 거래소는 노출되지 않음
currency=item.get("currency") or "USD",
)
def _to_stock_orders(self, item: dict, current_price: Optional[Decimal] = None) -> List[OverseasStockOrder]:
"""주문 응답을 OverseasStockOrder 목록으로 변환한다. CANCELED 주문은 [취소(CANCEL), 원주문(NEW)] 2건으로 분리한다."""
order = self._to_stock_order(item, current_price)
if item.get("status") != "CANCELED":
return [order]
cancel = self._to_stock_order(item, current_price)
cancel.order_no = None # 취소 요청 ID 미노출
cancel.req_type = OrderRequestType.CANCEL
cancel.org_order_no = order.order_no
cancel.price = Decimal(0) # KIS 취소 레코드와 동일하게 가격 없음
cancel.quantity = _num(max(Decimal(0), Decimal(order.quantity) - Decimal(order.filled_quantity))) # 취소된(미체결) 수량
cancel.filled_quantity = 0
cancel.filled_price = 0
# canceledAt 이 orderedAt 을 미러링하는 사례가 있어 실제 취소 시각이 아닐 수 있음
cancel_kr_time = _to_kr_time(item.get("canceledAt")) or order.order_kr_time
cancel.order_kr_time = cancel_kr_time
cancel.order_time = cancel_kr_time.astimezone(_NYT) if cancel_kr_time is not None else None
return [cancel, order]
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[OverseasStockOrder]:
"""주문 응답 목록을 현재가를 보강한 OverseasStockOrder 목록으로 변환한다. (취소된 주문은 2건으로 분리)"""
price_map = self._fetch_current_prices([item["symbol"] for item in items]) if items else {}
return [order for item in items for order in self._to_stock_orders(item, price_map.get(item["symbol"]))]
@staticmethod
def _orders_to_frame(orders: List[OverseasStockOrder]) -> 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"] = d["req_type"].name # NEW / CANCEL (분리된 취소 레코드)
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[OverseasStockOrder], pd.DataFrame]:
"""
미체결 주문을 조회합니다.
Args:
to_frame (bool): True 일 경우 order_no 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[OverseasStockOrder] | 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[OverseasStockOrder], pd.DataFrame]:
"""
미국 현지 기준 오늘(또는 지정 일자)의 주문 내역을 조회합니다.
Note:
취소된 주문은 취소(CANCEL) 레코드가 분리되어 원주문(NEW)과 함께 반환됩니다.
취소 레코드는 ``order_no=None`` 이며 ``org_order_no`` 로 원주문을 가리킵니다.
Args:
target_date (dtm.date): 조회할 미국 현지 기준 날짜 (기본: 오늘)
to_frame (bool): True 일 경우 order_no 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[OverseasStockOrder] | 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[OverseasStockOrder], pd.DataFrame]:
"""
주문 내역을 조회합니다.
Note:
취소된 주문은 취소(CANCEL) 레코드가 분리되어 원주문(NEW)과 함께 반환됩니다.
Args:
from_date (dtm.date): 조회 시작 날짜 (미국 현지 기준)
to_date (dtm.date): 조회 종료 날짜 (미국 현지 기준)
to_frame (bool): True 일 경우 order_no 를 인덱스로 한 DataFrame 으로 반환
Returns:
List[OverseasStockOrder] | 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[OverseasStockOrder]:
"""
주문 번호로 주문 정보를 조회합니다.
Note:
취소된 주문이라도 원주문 1건만 반환합니다. (취소 레코드 분리는 주문 내역 조회 전용)
Args:
order_no (str): 주문 번호
Returns:
OverseasStockOrder: 주문 정보. 찾을 수 없으면 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 시그니처 유지용