# -*- coding: utf-8 -*- """시세 수집 — 소스 하나에 기대지 않는다. 🔴 시세 제공자들은 데이터센터 IP를 막는다. 예외를 던지는 것이 아니라 빈 응답이나 봇 차단 페이지를 200으로 돌려주므로, 예외 처리만으로는 잡히지 않는다. 그래서 소스를 하나 두지 않고 사다리로 내려간다. 순서 1) Yahoo chart API 직접 호출 — yfinance 의 쿠키/크럼 절차를 건너뛴다 2) Stooq CSV — 미국 주식·ETF. 키가 필요 없다 3) CoinGecko — 코인 전용 🔑 "성공"의 판정은 예외 없음이 아니라 **막대 개수**다. 0건을 성공으로 세면 사다리가 첫 칸에서 멈춘다. """ import io, time import pandas as pd UA = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/122.0 Safari/537.36"} MIN_BARS = 30 # 이보다 적으면 실패로 본다 # 종목코드 → 소스별 심볼 MAP = { "NVDA": {"feed": "NVDA", "yahoo": "NVDA", "yfin": "NVDA", "stooq": "nvda.us"}, "GOLD": {"feed": "GOLD", "yahoo": "GLD", "yfin": "GLD", "stooq": "gld.us"}, "OIL": {"feed": "OIL", "yahoo": "USO", "yfin": "USO", "stooq": "uso.us"}, "BTC": {"feed": "BTC", "yahoo": "BTC-USD", "yfin": "BTC-USD", "stooq": "btcusd", "gecko": "bitcoin"}, } def _yahoo(sym, days): import requests u = ("https://query1.finance.yahoo.com/v8/finance/chart/%s" "?range=%s&interval=1d" % (sym, "10y" if days > 1200 else "5y")) r = requests.get(u, headers=UA, timeout=25) if r.status_code != 200: raise ValueError("HTTP %d %s" % (r.status_code, r.text[:60])) j = r.json()["chart"]["result"][0] ts = j["timestamp"] cl = j["indicators"]["quote"][0]["close"] # 수정주가가 있으면 그쪽을 쓴다. 분할·배당이 수익률에 섞이면 안 된다. adj = j.get("indicators", {}).get("adjclose") if adj and adj[0].get("adjclose"): cl = adj[0]["adjclose"] s = pd.Series(cl, index=pd.to_datetime(ts, unit="s")).dropna() s.index = s.index.normalize().tz_localize(None) return s[~s.index.duplicated(keep="last")] def _yfin(sym, days): """yfinance 자체 경로. 쿠키·크럼을 얻어 오는 절차가 따로 있어 결과가 다르다.""" import yfinance as yf d = yf.download(sym, period="5y", interval="1d", progress=False, auto_adjust=True, threads=False) if isinstance(d.columns, pd.MultiIndex): d.columns = d.columns.droplevel(1) s = d["Close"].dropna() s.index = pd.to_datetime(s.index).tz_localize(None) return s def _stooq(sym, days): import requests r = requests.get("https://stooq.com/q/d/l/?s=%s&i=d" % sym, headers=UA, timeout=25) if not r.text.lstrip().lower().startswith("date"): # 봇 차단 페이지가 200으로 온다. 200을 성공으로 세면 안 된다. raise ValueError("stooq 차단(HTML 응답)") d = pd.read_csv(io.StringIO(r.text)) if "Close" not in d.columns: raise ValueError("stooq 응답에 Close 없음") s = pd.Series(d["Close"].values, index=pd.to_datetime(d["Date"])).dropna() return s def _gecko(cid, days): import requests r = requests.get("https://api.coingecko.com/api/v3/coins/%s/market_chart" "?vs_currency=usd&days=%d&interval=daily" % (cid, min(days, 3650)), headers=UA, timeout=25) j = r.json() if "prices" not in j: raise ValueError("coingecko: %s" % str(j)[:120]) p = j["prices"] s = pd.Series([v for _, v in p], index=pd.to_datetime([t for t, _ in p], unit="ms")) s.index = s.index.normalize().tz_localize(None) return s[~s.index.duplicated(keep="last")] def _feed_h(sym, bars): """시간봉 피드. 채점 격자가 이걸 쓴다. 🔑 일봉으로 채점하면 개막일에 하루를 통째로 기다려야 첫 수익률이 나온다. 규칙에는 '매시간 갱신 가능'이라고 써 놓고 채점만 일 단위이면 어긋난다. """ import os from huggingface_hub import hf_hub_download repo = os.environ.get("FINCHAL_FEED", "FINAL-Bench/finchal-prices") tok = os.environ.get("HF_TOKEN") or os.environ.get("FINCHAL_HF_TOKEN") f = hf_hub_download(repo, "prices_1h/%s.csv" % sym, repo_type="dataset", token=tok) d = pd.read_csv(f) s = pd.Series(d["close"].values, index=pd.to_datetime(d["ts"])).dropna() s = s[~s.index.duplicated(keep="last")].sort_index() return s.iloc[-bars:] def hourly(asset, bars=2000): """시간봉. 피드에만 있다. 없으면 None 을 돌려주고 호출부가 일봉으로 내려간다.""" sym = MAP.get(asset, {}).get("feed") if not sym: return None try: s = _feed_h(sym, bars) return s if len(s) >= 24 else None except Exception: return None def _feed(sym, days): """우리가 발행한 시세 피드. 공개 소스가 전부 막힌 서버에서는 이게 유일한 길이다. 🔴 실측: HF Space에서 Yahoo 429 · Stooq 봇차단 · CoinGecko 차단. 바깥 인터넷은 되는데 시세 제공자들이 데이터센터 IP를 막는다. 그래서 시세를 '가져오는' 대신 '받는' 구조로 바꾼다 — 네트워크가 되는 곳에서 publish_prices.py 가 데이터셋에 올리고, 서버는 그 데이터셋을 읽는다. """ import os from huggingface_hub import hf_hub_download repo = os.environ.get("FINCHAL_FEED", "FINAL-Bench/finchal-prices") tok = os.environ.get("HF_TOKEN") or os.environ.get("FINCHAL_HF_TOKEN") f = hf_hub_download(repo, "prices/%s.csv" % sym, repo_type="dataset", token=tok) d = pd.read_csv(f) s = pd.Series(d["close"].values, index=pd.to_datetime(d["date"])).dropna() return s[~s.index.duplicated(keep="last")] # 피드를 맨 앞에 둔다. 서버에서 유일하게 통하는 경로이고, # 로컬에서는 피드가 없으면 바로 다음 칸으로 내려간다. LADDER = [("feed", _feed), ("yahoo", _yahoo), ("yfin", _yfin), ("stooq", _stooq), ("gecko", _gecko)] def daily(asset, days=1800): """일봉 종가. 어느 소스에서 왔는지도 함께 돌려준다. 반환: (시리즈, 소스명, 시도기록) """ m = MAP.get(asset, {}) trail = [] for name, fn in LADDER: sym = m.get(name) if not sym: continue t0 = time.time() try: s = fn(sym, days) n = len(s) trail.append("%s:%d봉/%.1fs" % (name, n, time.time() - t0)) if n >= MIN_BARS: # 0건은 성공이 아니다 return s.sort_index().iloc[-days:], name, trail except Exception as e: # 🔴 예외 '종류'만 남기면 왜 막혔는지 모른다. 진단이 안 되는 로그는 # 없는 로그와 같다. 메시지까지 남긴다. trail.append("%s:%s %s" % (name, type(e).__name__, str(e)[:90])) return None, None, trail if __name__ == "__main__": for a in MAP: s, src, tr = daily(a, 400) print("%-5s %-7s %s %s" % ( a, src or "실패", ("%d봉 마지막 %.2f" % (len(s), s.iloc[-1])) if s is not None else "-", " · ".join(tr)))