app/fetch.py

# -*- 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)))