"""Exchange adapters — auto-selects first reachable source. Priority: BingX (user's exchange) → Binance → Bybit → Kraken USDT → Kraken USD. """ from __future__ import annotations import time, warnings import pandas as pd import requests warnings.filterwarnings("ignore", message="urllib3 v2 only supports OpenSSL") KLINE_LIMIT = 500 INTERVAL_MS = {"1h": 3_600_000, "15m": 900_000, "5m": 300_000} _session = requests.Session() _session.headers["User-Agent"] = "trade-copilot/1.0" def _get(url: str, params: dict | None = None): for attempt in range(3): try: r = _session.get(url, params=params, timeout=15) if r.status_code == 429: time.sleep(2 ** (attempt + 1)) continue r.raise_for_status() return r.json() except requests.RequestException: if attempt == 2: raise time.sleep(1.5 * (attempt + 1)) # ── BingX USDT-M perpetuals (user's exchange — top priority) ────────────── class BingX: name = "bingx" market_note = "USDT-M perpetual futures (BingX)" anchors = ["BTC-USDT", "ETH-USDT"] min_vol = 5_000_000 _iv = {"1h": "1h", "15m": "15m", "5m": "5m"} _base = "https://open-api.bingx.com" @staticmethod def _ts(): return str(int(time.time() * 1000)) def _get(self, path, params=None): p = params or {} p["timestamp"] = self._ts() return _get(self._base + path, p) @staticmethod def _ok(r): if isinstance(r, dict) and r.get("code", 0) != 0: raise RuntimeError(r.get("msg", "BingX error")) return r def ping(self): r = self._get("/openApi/swap/v2/quote/ticker", {"symbol": "BTC-USDT"}) self._ok(r) def universe(self): r = self._ok(self._get("/openApi/swap/v2/quote/contracts")) return {c["symbol"] for c in r.get("data", []) if c.get("symbol", "").endswith("-USDT")} def tickers(self): r = self._ok(self._get("/openApi/swap/v2/quote/ticker")) out = [] for t in (r.get("data") or []): sym = t.get("symbol", "") if not sym.endswith("-USDT"): continue try: last = float(t["lastPrice"]) open_ = float(t["openPrice"]) high = float(t["highPrice"]) low = float(t["lowPrice"]) vol = float(t["volume"]) # base volume qv = float(t.get("quoteVolume") or vol * last) out.append({ "symbol": sym, "quote_volume_usd": qv, "change_24h_pct": (last - open_) / open_ * 100 if open_ > 0 else 0, "high": high, "low": low, "last_price": last }) except Exception: pass return out def funding(self, symbol: str) -> float | None: try: r = self._ok(self._get("/openApi/swap/v2/quote/premiumIndex", {"symbol": symbol})) return float((r.get("data") or {}).get("lastFundingRate", 0)) except Exception: return None def open_interest(self, symbol: str) -> float | None: try: r = self._ok(self._get("/openApi/swap/v2/quote/openInterest", {"symbol": symbol})) return float((r.get("data") or {}).get("openInterest", 0)) except Exception: return None def taker_ratio(self, symbol: str, period: str = "5m") -> float | None: """Taker buy/sell ratio from BingX dedicated endpoint. Returns float in [0,1] where >0.5 = more buy-side aggression, or None on error. period: "5m" | "15m" | "30m" | "1h" — lookback window. """ try: r = self._ok(self._get("/openApi/swap/v2/quote/takerLongShortRatio", { "symbol": symbol, "period": period, "limit": "1", })) data = r.get("data") or [] if not data: return None entry = data[0] if isinstance(data, list) else data buy_vol = float(entry.get("buyVol", entry.get("takerBuyVol", 0)) or 0) sell_vol = float(entry.get("sellVol", entry.get("takerSellVol", 0)) or 0) total = buy_vol + sell_vol if total <= 0: return None return round(buy_vol / total, 4) except Exception: return None def klines(self, symbol: str, interval: str) -> pd.DataFrame: r = self._ok(self._get("/openApi/swap/v3/quote/klines", { "symbol": symbol, "interval": self._iv[interval], "limit": str(KLINE_LIMIT) })) rows = r.get("data") or [] if not rows: raise RuntimeError(f"No kline data for {symbol}") df = pd.DataFrame(rows) # BingX v3 returns: {"time", "open", "high", "low", "close", "volume"} # Some contract types also include takerBuyBaseAssetVolume or takerBuyVolume. df = df.rename(columns={"time": "open_time"}) df["open_time"] = df["open_time"].astype("int64") for c in ("open", "high", "low", "close", "volume"): df[c] = df[c].astype(float) # Normalise taker buy volume — try known BingX field names, NaN if absent _tbv_candidates = ("takerBuyBaseAssetVolume", "takerBuyVolume", "taker_buy_vol") _tbv_col = next((c for c in _tbv_candidates if c in df.columns), None) df["taker_buy_vol"] = ( pd.to_numeric(df[_tbv_col], errors="coerce") if _tbv_col is not None else float("nan") ) df = df.sort_values("open_time").reset_index(drop=True) now_ms = int(time.time() * 1000) if int(df["open_time"].iloc[-1]) + INTERVAL_MS[interval] > now_ms: df = df.iloc[:-1] return df[["open_time", "open", "high", "low", "close", "volume", "taker_buy_vol"]].reset_index(drop=True) def tv_symbol(self, sym): # BTC-USDT → BINGX:BTCUSDT.P return "BINGX:" + sym.replace("-", "") + ".P" # ── Binance USDT-M perpetuals ────────────────────────────────────────────── class Binance: name = "binance" market_note = "USDT perpetual futures" anchors = ["BTCUSDT", "ETHUSDT"] min_vol = 75_000_000 def ping(self): _get("https://fapi.binance.com/fapi/v1/ping") def universe(self): return {s["symbol"] for s in _get("https://fapi.binance.com/fapi/v1/exchangeInfo")["symbols"] if s["contractType"] == "PERPETUAL" and s["status"] == "TRADING" and s["quoteAsset"] == "USDT"} def tickers(self): out = [] for t in _get("https://fapi.binance.com/fapi/v1/ticker/24hr"): try: out.append({"symbol": t["symbol"], "quote_volume_usd": float(t["quoteVolume"]), "change_24h_pct": float(t["priceChangePercent"]), "high": float(t["highPrice"]), "low": float(t["lowPrice"]), "last_price": float(t["lastPrice"])}) except (KeyError, ValueError): pass return out def funding(self, symbol: str) -> float | None: try: data = _get("https://fapi.binance.com/fapi/v1/premiumIndex", {"symbol": symbol}) return float(data["lastFundingRate"]) except Exception: return None def open_interest(self, symbol: str) -> float | None: try: data = _get("https://fapi.binance.com/fapi/v1/openInterest", {"symbol": symbol}) return float(data["openInterest"]) except Exception: return None def klines(self, symbol: str, interval: str) -> pd.DataFrame: raw = _get("https://fapi.binance.com/fapi/v1/klines", {"symbol": symbol, "interval": interval, "limit": KLINE_LIMIT}) df = pd.DataFrame(raw, columns=["open_time","open","high","low","close", "volume","close_time","qv","trades","tb","tq","ig"]) for c in ("open","high","low","close","volume"): df[c] = df[c].astype(float) # Binance column "tb" = takerBuyBaseAssetVolume — normalise to taker_buy_vol df["taker_buy_vol"] = pd.to_numeric(df["tb"], errors="coerce") now_ms = int(time.time() * 1000) if int(df["close_time"].iloc[-1]) > now_ms: df = df.iloc[:-1] return df[["open_time","open","high","low","close","volume", "taker_buy_vol"]].reset_index(drop=True) def tv_symbol(self, sym): return f"BINANCE:{sym}.P" # ── Bybit linear perpetuals ──────────────────────────────────────────────── class Bybit: name = "bybit" market_note = "USDT linear perpetuals" anchors = ["BTCUSDT", "ETHUSDT"] min_vol = 75_000_000 _iv = {"1h": "60", "15m": "15", "5m": "5"} @staticmethod def _ok(r): if r.get("retCode") != 0: raise RuntimeError(r.get("retMsg")) return r["result"] def ping(self): self._ok(_get("https://api.bybit.com/v5/market/time")) def universe(self): res = self._ok(_get("https://api.bybit.com/v5/market/instruments-info", {"category": "linear", "limit": 1000})) return {s["symbol"] for s in res["list"] if s.get("status") == "Trading" and s.get("quoteCoin") == "USDT" and s.get("contractType") == "LinearPerpetual"} def tickers(self): res = self._ok(_get("https://api.bybit.com/v5/market/tickers", {"category": "linear"})) out = [] for t in res["list"]: try: out.append({"symbol": t["symbol"], "quote_volume_usd": float(t["turnover24h"]), "change_24h_pct": float(t["price24hPcnt"]) * 100, "high": float(t["highPrice24h"]), "low": float(t["lowPrice24h"]), "last_price": float(t["lastPrice"])}) except (KeyError, ValueError): pass return out def funding(self, symbol): return None def open_interest(self, symbol): return None def klines(self, symbol: str, interval: str) -> pd.DataFrame: res = self._ok(_get("https://api.bybit.com/v5/market/kline", {"category": "linear", "symbol": symbol, "interval": self._iv[interval], "limit": KLINE_LIMIT})) df = pd.DataFrame(list(reversed(res["list"])), columns=["open_time","open","high","low","close","volume","turnover"]) df["open_time"] = df["open_time"].astype("int64") for c in ("open","high","low","close","volume"): df[c] = df[c].astype(float) now_ms = int(time.time() * 1000) if int(df["open_time"].iloc[-1]) + INTERVAL_MS[interval] > now_ms: df = df.iloc[:-1] return df[["open_time","open","high","low","close","volume"]].reset_index(drop=True) def tv_symbol(self, sym): return f"BYBIT:{sym}.P" # ── Kraken USDT pairs (US & India accessible) ───────────────────────────── class KrakenUSDT: """Kraken spot pairs quoted in USDT — symbols like BTCUSDT, ETHUSDT.""" name = "kraken" market_note = "USDT spot pairs (US & India accessible)" anchors = ["BTCUSDT", "ETHUSDT"] min_vol = 1_000_000 _iv = {"1h": 60, "15m": 15, "5m": 5} def __init__(self): self._alt_to_key: dict[str, str] = {} @staticmethod def _ok(r): if r.get("error"): raise RuntimeError(r["error"]) return r["result"] def ping(self): self._ok(_get("https://api.kraken.com/0/public/Time")) def _load_pairs(self): if self._alt_to_key: return res = self._ok(_get("https://api.kraken.com/0/public/AssetPairs")) for key, p in res.items(): ws = p.get("wsname", "") # e.g. "BTC/USDT" alt = p.get("altname", "") # e.g. "XBTUSDT" if ws.endswith("/USDT") and not alt.endswith(".d"): # normalise XBT→BTC in the symbol we expose clean = ws.replace("XBT/", "BTC/").replace("/", "") # "BTCUSDT" self._alt_to_key[clean] = key def universe(self): self._load_pairs(); return set(self._alt_to_key) def tickers(self): self._load_pairs() key_to_clean = {v: k for k, v in self._alt_to_key.items()} res = self._ok(_get("https://api.kraken.com/0/public/Ticker")) out = [] for key, t in res.items(): clean = key_to_clean.get(key) if not clean: continue try: last = float(t["c"][0]); open_ = float(t["o"]) vol24 = float(t["v"][1]); vwap24 = float(t["p"][1]) out.append({"symbol": clean, "quote_volume_usd": vol24 * vwap24, "change_24h_pct": (last - open_) / open_ * 100 if open_ > 0 else 0, "high": float(t["h"][1]), "low": float(t["l"][1]), "last_price": last}) except Exception: pass return out def funding(self, symbol): return None def open_interest(self, symbol): return None def klines(self, symbol: str, interval: str) -> pd.DataFrame: self._load_pairs() key = self._alt_to_key.get(symbol, symbol) res = self._ok(_get("https://api.kraken.com/0/public/OHLC", {"pair": key, "interval": self._iv[interval]})) rows = next(v for k, v in res.items() if k != "last") df = pd.DataFrame(rows, columns=["t","open","high","low","close","vwap","volume","count"]) df["open_time"] = df["t"].astype("int64") * 1000 for c in ("open","high","low","close","volume"): df[c] = df[c].astype(float) now_ms = int(time.time() * 1000) if int(df["open_time"].iloc[-1]) + INTERVAL_MS[interval] > now_ms: df = df.iloc[:-1] return df[["open_time","open","high","low","close","volume"]].reset_index(drop=True) def tv_symbol(self, sym): return "KRAKEN:" + sym # e.g. KRAKEN:BTCUSDT # ── Kraken spot USD fallback ─────────────────────────────────────────────── class Kraken: name = "kraken_usd" market_note = "spot USD pairs (fallback)" anchors = ["XBTUSD", "ETHUSD"] min_vol = 10_000_000 _iv = {"1h": 60, "15m": 15, "5m": 5} def __init__(self): self._alt_to_key: dict[str, str] = {} @staticmethod def _ok(r): if r.get("error"): raise RuntimeError(r["error"]) return r["result"] def ping(self): self._ok(_get("https://api.kraken.com/0/public/Time")) def _load_pairs(self): if self._alt_to_key: return res = self._ok(_get("https://api.kraken.com/0/public/AssetPairs")) for key, p in res.items(): ws = p.get("wsname", "") alt = p.get("altname", "") if ws.endswith("/USD") and not alt.endswith(".d"): self._alt_to_key[alt] = key def universe(self): self._load_pairs(); return set(self._alt_to_key) def tickers(self): self._load_pairs() key_to_alt = {v: k for k, v in self._alt_to_key.items()} res = self._ok(_get("https://api.kraken.com/0/public/Ticker")) out = [] for key, t in res.items(): alt = key_to_alt.get(key) if not alt: continue try: last = float(t["c"][0]); open_ = float(t["o"]) vol24 = float(t["v"][1]); vwap24 = float(t["p"][1]) out.append({"symbol": alt, "quote_volume_usd": vol24 * vwap24, "change_24h_pct": (last - open_) / open_ * 100 if open_ > 0 else 0, "high": float(t["h"][1]), "low": float(t["l"][1]), "last_price": last}) except Exception: pass return out def funding(self, symbol): return None def open_interest(self, symbol): return None def klines(self, symbol: str, interval: str) -> pd.DataFrame: self._load_pairs() key = self._alt_to_key.get(symbol, symbol) res = self._ok(_get("https://api.kraken.com/0/public/OHLC", {"pair": key, "interval": self._iv[interval]})) rows = next(v for k, v in res.items() if k != "last") df = pd.DataFrame(rows, columns=["t","open","high","low","close","vwap","volume","count"]) df["open_time"] = df["t"].astype("int64") * 1000 for c in ("open","high","low","close","volume"): df[c] = df[c].astype(float) now_ms = int(time.time() * 1000) if int(df["open_time"].iloc[-1]) + INTERVAL_MS[interval] > now_ms: df = df.iloc[:-1] return df[["open_time","open","high","low","close","volume"]].reset_index(drop=True) def tv_symbol(self, sym): return "KRAKEN:" + sym.replace("XBT", "BTC") # Priority: BingX → Binance → Bybit → Kraken USDT spot → Kraken USD spot SOURCES: list = [BingX(), Binance(), Bybit(), KrakenUSDT(), Kraken()] def pick_source(forced: str | None = None): targets = SOURCES if not forced or forced == "auto" \ else [s for s in SOURCES if s.name == forced] errors = [] for src in targets: try: src.ping() return src except Exception as e: msg = str(e).split("for url")[0].strip()[:80] errors.append(f" {src.name}: {msg}") raise RuntimeError("No exchange reachable:\n" + "\n".join(errors))