exchange.py

"""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))