1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204 | """
토스증권 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())
|