eink-push/btc_api.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
比特币 K 线行情客户端 —— 免费、无需密钥的公开接口，按顺序尝试多个源，首个成功即返回。

源（均为公开 REST，无鉴权）：
  1. OKX      GET /api/v5/market/candles?instId=BTC-USDT&bar=<1H|1D>       —— 主源，本地直连可达、数据最新
  2. Coinbase GET /products/BTC-USD/candles?granularity=<3600|86400>        —— 备源（USD 计价，与 USDT 数值几乎一致）
  3. Huobi    GET /market/history/kline?period=<60min|1day>&symbol=btcusdt  —— 备源
  4. Gate.io  GET /api/v4/spot/candlesticks?currency_pair=BTC_USDT&interval=<1h|1d>
  5. Huobi AWS 同 3，域名 api-aws.huobi.pro
  6. Binance  GET data-api.binance.vision/api/v3/klines?symbol=BTCUSDT&interval=<1h|1d>（公开行情镜像域名）
  （api.binance.com 在本地区返回 restricted location，不列入。）

顺序可用 BTC_SOURCES（逗号分隔源名）覆盖：境内服务器上 okx/coinbase/huobi 主域名均不通且逐个超时（合计 >15s，
超出 FETCH_BUDGET），须把 gateio/huobi_aws/binance_vision 排在前面（2026-09-18 部署 NGCQ 实测）。

周期用统一代号 bar："1H"（时 K）/ "1D"（日 K），由 BARS 映射到各源的写法。
统一输出：按时间**升序**的 list[dict]，每根 {"ts": epoch 秒, "o", "h", "l", "c"}（float），最后一根为当前未收盘的那根。
任何源失败抛异常并尝试下一个；全部失败抛 RuntimeError，由 data 层决定沿用缓存或占位。
"""
import json
import urllib.request

TIMEOUT = 8
UA = "eink-push/1.0 (+https://github.com/)"

# 周期代号 → 各源参数：(OKX bar, Coinbase granularity 秒, Huobi period, Gate.io/Binance interval)
BARS = {"1H": ("1H", 3600, "60min", "1h"), "1D": ("1D", 86400, "1day", "1d")}


def _get_json(url):
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
        return json.loads(resp.read().decode("utf-8"))


def _okx(bar, limit):
    j = _get_json(f"https://www.okx.com/api/v5/market/candles?instId=BTC-USDT&bar={BARS[bar][0]}&limit={limit}")
    if str(j.get("code")) != "0":
        raise RuntimeError(f"okx code={j.get('code')} msg={j.get('msg')}")
    rows = j["data"]  # 最新在前：[ts_ms, o, h, l, c, vol, ...]
    out = [{"ts": int(r[0]) // 1000, "o": float(r[1]), "h": float(r[2]), "l": float(r[3]), "c": float(r[4])} for r in rows]
    return sorted(out, key=lambda x: x["ts"])


def _coinbase(bar, limit):
    rows = _get_json(f"https://api.exchange.coinbase.com/products/BTC-USD/candles?granularity={BARS[bar][1]}")
    # 最新在前：[time, low, high, open, close, volume]
    out = [{"ts": int(r[0]), "o": float(r[3]), "h": float(r[2]), "l": float(r[1]), "c": float(r[4])} for r in rows]
    return sorted(out, key=lambda x: x["ts"])[-limit:]


def _huobi(bar, limit, host="api.huobi.pro"):
    j = _get_json(f"https://{host}/market/history/kline?period={BARS[bar][2]}&size={limit}&symbol=btcusdt")
    if j.get("status") != "ok":
        raise RuntimeError(f"huobi status={j.get('status')} err={j.get('err-msg')}")
    out = [{"ts": int(r["id"]), "o": float(r["open"]), "h": float(r["high"]), "l": float(r["low"]), "c": float(r["close"])}
           for r in j["data"]]
    return sorted(out, key=lambda x: x["ts"])


def _huobi_aws(bar, limit):
    return _huobi(bar, limit, host="api-aws.huobi.pro")


def _gateio(bar, limit):
    rows = _get_json(f"https://api.gateio.ws/api/v4/spot/candlesticks?currency_pair=BTC_USDT&interval={BARS[bar][3]}&limit={limit}")
    # 升序：[ts 秒, 计价币成交额, close, high, low, open, 基础币成交量, 是否收盘]
    out = [{"ts": int(r[0]), "o": float(r[5]), "h": float(r[3]), "l": float(r[4]), "c": float(r[2])} for r in rows]
    return sorted(out, key=lambda x: x["ts"])


def _binance_vision(bar, limit):
    rows = _get_json(f"https://data-api.binance.vision/api/v3/klines?symbol=BTCUSDT&interval={BARS[bar][3]}&limit={limit}")
    # 升序：[openTime ms, o, h, l, c, vol, closeTime, ...]
    out = [{"ts": int(r[0]) // 1000, "o": float(r[1]), "h": float(r[2]), "l": float(r[3]), "c": float(r[4])} for r in rows]
    return sorted(out, key=lambda x: x["ts"])


SOURCES = (("okx", _okx), ("coinbase", _coinbase), ("huobi", _huobi),
           ("gateio", _gateio), ("huobi_aws", _huobi_aws), ("binance_vision", _binance_vision))
_BY_NAME = dict(SOURCES)


def _ordered_sources():
    """按 config.BTC_SOURCES（逗号分隔源名）排序；未列出的源按默认顺序追加到末尾，未知名字忽略。"""
    try:
        import config
        wanted = [s.strip() for s in config.BTC_SOURCES.split(",") if s.strip()]
    except Exception:  # noqa: BLE001 —— 单独运行本模块或 config 缺失时用默认顺序
        wanted = []
    ordered = [(n, _BY_NAME[n]) for n in wanted if n in _BY_NAME]
    ordered += [(n, f) for n, f in SOURCES if n not in dict(ordered)]
    return ordered


def get_candles(bar="1H", limit=36):
    """返回 (源名, 升序 K 线列表)。bar 为 "1H"/"1D"。逐源尝试，全部失败抛 RuntimeError。"""
    if bar not in BARS:
        raise ValueError(f"不支持的周期 {bar!r}，可选：{list(BARS)}")
    errors = []
    for name, fn in _ordered_sources():
        try:
            rows = fn(bar, limit)
            if len(rows) >= 2:
                return name, rows[-limit:]
            errors.append(f"{name}: 数据不足({len(rows)})")
        except Exception as e:  # noqa: BLE001 —— 逐源兜底，任何异常都换下一个源
            errors.append(f"{name}: {e}")
    raise RuntimeError(f"BTC {bar} K 线全部源失败：" + "；".join(errors))


if __name__ == "__main__":
    src, rows = get_candles("1H", 5)
    print(src)
    for r in rows:
        print(r)