""" 토스증권 Open API 공용 클라이언트. - OAuth2 client_credentials 토큰 발급 + 24h 캐싱(~/.cache/toss/token.json). ※ 토스는 client 당 유효 토큰 1개 — 재발급 시 이전 토큰 무효화되므로 캐시 재사용이 중요. - 시세(candles/prices/orderbook/price-limits), 종목정보(stocks/warnings), 시장정보(exchange-rate/market-calendar) 래퍼. - yahoo_chart_client.yc_close_wide 대체용 `close_wide()` 제공 (KR+US 통합). 자격증명: 환경변수 TOSS_INVEST_CLIENT_ID / TOSS_INVEST_CLIENT_SECRET (없으면 ~/.env 파싱). """ from __future__ import annotations import json, os, time, threading from datetime import datetime from pathlib import Path from typing import Any import requests BASE = "https://openapi.tossinvest.com" _TOKEN_CACHE = Path(os.path.expanduser("~/.cache/toss/token.json")) _lock = threading.Lock() def _load_env(key: str) -> str | None: v = os.environ.get(key) if v: return v envp = Path(os.path.expanduser("~/.env")) if envp.exists(): for line in envp.read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith(key + "="): return line.split("=", 1)[1].strip().strip('"').strip("'") return None def _to_toss_symbol(ticker: str) -> str: """yfinance/내부 표기 → 토스 심볼. '005930.KS'/'005930.KQ' → '005930', 'AAPL' → 'AAPL'.""" t = str(ticker).strip().upper() for suf in (".KS", ".KQ", ".KR"): if t.endswith(suf): return t[: -len(suf)] # 지수(^KS11 등)는 토스 미지원 → 그대로 반환(호출측에서 거름) return t class TossClient: def __init__(self, client_id: str | None = None, client_secret: str | None = None, timeout: int = 20): self.client_id = client_id or _load_env("TOSS_INVEST_CLIENT_ID") self.client_secret = client_secret or _load_env("TOSS_INVEST_CLIENT_SECRET") if not (self.client_id and self.client_secret): raise RuntimeError("TOSS_INVEST_CLIENT_ID/SECRET 없음 (~/.env 확인)") self.timeout = timeout self._sess = requests.Session() # ---- auth ------------------------------------------------------------- def _cached_token(self) -> str | None: try: d = json.loads(_TOKEN_CACHE.read_text()) if d.get("expires_at", 0) - 120 > time.time(): # 만료 2분 전까지 재사용 return d["access_token"] except Exception: pass return None def token(self) -> str: with _lock: tok = self._cached_token() if tok: return tok r = self._sess.post( f"{BASE}/oauth2/token", data={"grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret}, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=self.timeout, ) r.raise_for_status() d = r.json() _TOKEN_CACHE.parent.mkdir(parents=True, exist_ok=True) _TOKEN_CACHE.write_text(json.dumps({ "access_token": d["access_token"], "expires_at": time.time() + int(d.get("expires_in", 86400)), })) try: os.chmod(_TOKEN_CACHE, 0o600) except Exception: pass return d["access_token"] # ---- core request (429 재시도) --------------------------------------- def _get(self, path: str, params: dict | None = None, _retry: int = 3) -> Any: for attempt in range(_retry): r = self._sess.get( f"{BASE}{path}", params=params or {}, headers={"Authorization": f"Bearer {self.token()}", "Accept": "application/json"}, timeout=self.timeout, ) if r.status_code == 429: wait = int(r.headers.get("Retry-After", "1")) time.sleep(min(wait, 5)) continue if r.status_code == 401: # 토큰 문제 → 캐시 무효화 후 1회 재발급 _TOKEN_CACHE.unlink(missing_ok=True) if attempt < _retry - 1: continue r.raise_for_status() d = r.json() return d.get("result", d) r.raise_for_status() return None # ---- market data ------------------------------------------------------ def candles(self, symbol: str, interval: str = "1d", count: int = 100, before: str | None = None, adjusted: bool = True) -> dict: p = {"symbol": _to_toss_symbol(symbol), "interval": interval, "count": count, "adjusted": str(adjusted).lower()} if before: p["before"] = before return self._get("/api/v1/candles", p) def prices(self, symbols: list[str] | str) -> list[dict]: if isinstance(symbols, (list, tuple)): symbols = ",".join(_to_toss_symbol(s) for s in symbols) return self._get("/api/v1/prices", {"symbols": symbols}) def orderbook(self, symbol: str) -> dict: return self._get("/api/v1/orderbook", {"symbol": _to_toss_symbol(symbol)}) def price_limits(self, symbol: str) -> dict: return self._get("/api/v1/price-limits", {"symbol": _to_toss_symbol(symbol)}) def stocks(self, symbols: list[str] | str) -> list[dict]: if isinstance(symbols, (list, tuple)): symbols = ",".join(_to_toss_symbol(s) for s in symbols) return self._get("/api/v1/stocks", {"symbols": symbols}) def warnings(self, symbol: str) -> list[dict]: return self._get(f"/api/v1/stocks/{_to_toss_symbol(symbol)}/warnings") def exchange_rate(self, base: str = "USD", quote: str = "KRW") -> dict: return self._get("/api/v1/exchange-rate", {"baseCurrency": base, "quoteCurrency": quote}) def market_calendar(self, country: str = "KR", date: str | None = None) -> dict: p = {"date": date} if date else {} return self._get(f"/api/v1/market-calendar/{country}", p) def candles_paged(self, symbol: str, interval: str = "1d", total: int = 200, adjusted: bool = True) -> list[dict]: """nextBefore 페이지네이션으로 최대 `total` 봉 수집 (호출당 200 한도 우회).""" out: list[dict] = [] before = None while len(out) < total: res = self.candles(symbol, interval=interval, count=min(200, total - len(out)), before=before, adjusted=adjusted) rows = res.get("candles", []) if isinstance(res, dict) else [] if not rows: break out.extend(rows) before = res.get("nextBefore") if not before: break return out # ---- yahoo_chart_client.yc_close_wide 대체 ---------------------------- def close_wide(self, tickers: list[str], count: int = 200, interval: str = "1d"): """종목별 종가 wide DataFrame (index=날짜, columns=원본 ticker). 실패 종목은 컬럼 제외.""" import pandas as pd series = {} for tk in tickers: sym = _to_toss_symbol(tk) if sym.startswith("^") or not sym.replace(".", "").replace("-", "").isalnum(): continue # 지수 등 미지원 스킵 try: rows = self.candles_paged(sym, interval=interval, total=count) if not rows: continue idx = [pd.Timestamp(c["timestamp"]).tz_localize(None).normalize() for c in rows] vals = [float(c["closePrice"]) for c in rows] s = pd.Series(vals, index=idx).sort_index() series[tk] = s[~s.index.duplicated(keep="last")] except Exception: continue if not series: return pd.DataFrame() return pd.DataFrame(series).sort_index() _default: TossClient | None = None def get_client() -> TossClient: global _default if _default is None: _default = TossClient() return _default if __name__ == "__main__": c = get_client() print("token ok:", bool(c.token())) df = c.close_wide(["005930", "000660.KS", "AAPL"], count=5) print(df.tail())