"""

Market Data -- Multi-exchange market data with automatic failover.

Fetches open interest, funding rates, and long/short ratios from

Hyperliquid, Binance, Bybit, and OKX. Automatically falls through

to the next source on failure.

Usage::

from market_data import MarketData

md = MarketData()

price, source = md.get_price_with_failover("BTC")

oi = md.get_btc_open_interest()

funding = md.get_funding_rates()

snapshot = md.get_full_snapshot("BTC")

"""

from __future__ import annotations

import argparse

import json

import time

from datetime import UTC, datetime

import requests

class DataSourceManager:

"""Tracks health and latency across data sources for failover decisions."""

SOURCES = ["hyperliquid", "binance", "bybit"]

def __init__(self) -> None:

self.current_source: str = "hyperliquid"

self.latencies: dict[str, float] = {}

self.failures: dict[str, int] = {}

def record_success(self, source: str, latency_ms: float) -> None:

self.latencies[source] = latency_ms

self.failures[source] = 0

self.current_source = source

def record_failure(self, source: str) -> None:

self.failures[source] = self.failures.get(source, 0) + 1

def get_next_source(self, failed_source: str) -> str | None:

"""Return the next source to try after *failed_source*, or ``None``."""

idx = self.SOURCES.index(failed_source) if failed_source in self.SOURCES else -1

for i in range(idx + 1, len(self.SOURCES)):

return self.SOURCES[i]

return None

def get_status(self) -> dict:

return {

"current_source": self.current_source,

"latencies": self.latencies,

"failures": self.failures,

"available_sources": self.SOURCES,

}

class MarketData:

"""Multi-exchange market data with automatic failover.

Aggregates data from Hyperliquid, Binance, Bybit, and OKX. Each

fetch method tries sources in order and records latency / failures

for observability.

"""

def __init__(self) -> None:

self.hyperliquid = "https://api.hyperliquid.xyz/info"

self.binance_futures = "https://fapi.binance.com/fapi/v1"

self.bybit = "https://api.bybit.com/v5/market"

self.okx = "https://www.okx.com/api/v5/public"

self.source_manager = DataSourceManager()

# -- price with failover ------------------------------------------------

def get_price_with_failover(self, symbol: str = "BTC") -> tuple[float, str]:

"""Get a price with automatic failover across exchanges.

Returns ``(price, source_name)``. If all sources fail, returns

``(0, "none")``.

"""

sources = [

("hyperliquid", self._get_hyperliquid_price),

("binance", self._get_binance_price),

("bybit", self._get_bybit_price),

]

for source_name, fetch_func in sources:

try:

start = time.time()

price = fetch_func(symbol)

latency = (time.time() - start) * 1000

if price and price > 0:

self.source_manager.record_success(source_name, latency)

return price, source_name

except Exception:

self.source_manager.record_failure(source_name)

continue

return 0, "none"

def _get_hyperliquid_price(self, symbol: str) -> float:

for attempt in range(3):

r = requests.post(self.hyperliquid, json={"type": "allMids"}, timeout=5)

if r.status_code == 429:

time.sleep(min(2 ** attempt * 2, 8))

continue

data = r.json()

return float((data or {}).get(symbol, 0))

return 0

def _get_binance_price(self, symbol: str) -> float:

r = requests.get(

f"{self.binance_futures}/ticker/price?symbol={symbol}USDT", timeout=5

)

return float(r.json()["price"])

def _get_bybit_price(self, symbol: str) -> float:

r = requests.get(

f"{self.bybit}/tickers?category=linear&symbol={symbol}USDT", timeout=5

)

data = r.json()

if data["retCode"] == 0:

return float(data["result"]["list"][0]["lastPrice"])

return 0

# -- status -------------------------------------------------------------

def get_data_source_status(self) -> dict:

"""Current health/latency status for all data sources."""

return self.source_manager.get_status()

# -- open interest ------------------------------------------------------

def get_btc_open_interest(self) -> dict:

"""Aggregate BTC open interest from Binance, Bybit, and OKX.

Returns a dict keyed by exchange name, plus ``total_oi_btc`` and

``fetched_at``.

"""

oi_data: dict = {}

# Binance

try:

r = requests.get(

f"{self.binance_futures}/openInterest?symbol=BTCUSDT", timeout=5

)

data = r.json()

oi_data["binance"] = {

"oi_btc": float(data["openInterest"]),

"timestamp": data["time"],

}

except Exception as e:

oi_data["binance"] = {"error": str(e)}

# Bybit

try:

r = requests.get(

f"{self.bybit}/tickers?category=linear&symbol=BTCUSDT", timeout=5

)

data = r.json()

if data["retCode"] == 0:

ticker = data["result"]["list"][0]

oi_data["bybit"] = {

"oi_btc": float(ticker["openInterest"]),

"oi_value": float(ticker["openInterestValue"]),

"funding_rate": float(ticker["fundingRate"]),

"next_funding": ticker["nextFundingTime"],

}

except Exception as e:

oi_data["bybit"] = {"error": str(e)}

# OKX

try:

r = requests.get(

f"{self.okx}/open-interest?instType=SWAP&instId=BTC-USDT-SWAP",

timeout=5,

)

data = r.json()

if data["code"] == "0":

oi_data["okx"] = {

"oi_btc": float(data["data"][0]["oiCcy"]),

"oi_value": float(data["data"][0]["oiUsd"]),

"timestamp": data["data"][0]["ts"],

}

except Exception as e:

oi_data["okx"] = {"error": str(e)}

total_oi = sum(

d["oi_btc"] for d in oi_data.values() if isinstance(d, dict) and "oi_btc" in d

)

oi_data["total_oi_btc"] = total_oi

oi_data["fetched_at"] = datetime.now(UTC).isoformat()

return oi_data

# -- funding rates ------------------------------------------------------

def get_funding_rates(self) -> dict:

"""Aggregate BTC funding rates from Binance, Bybit, and OKX.

Returns per-exchange rates plus an ``average`` entry.

"""

funding: dict = {}

# Binance

try:

r = requests.get(

f"{self.binance_futures}/fundingRate?symbol=BTCUSDT&limit=1", timeout=5

)

data = r.json()[0]

rate = float(data["fundingRate"])

funding["binance"] = {

"rate": rate,

"rate_pct": rate * 100,

"annualized": rate * 3 * 365 * 100, # 8-hour funding

"next_funding": data["fundingTime"],

}

except Exception as e:

funding["binance"] = {"error": str(e)}

# Bybit

try:

r = requests.get(

f"{self.bybit}/tickers?category=linear&symbol=BTCUSDT", timeout=5

)

data = r.json()

if data["retCode"] == 0:

rate = float(data["result"]["list"][0]["fundingRate"])

funding["bybit"] = {

"rate": rate,

"rate_pct": rate * 100,

"annualized": rate * 3 * 365 * 100,

}

except Exception as e:

funding["bybit"] = {"error": str(e)}

# OKX

try:

r = requests.get(

f"{self.okx}/funding-rate?instId=BTC-USDT-SWAP", timeout=5

)

data = r.json()

if data["code"] == "0":

rate = float(data["data"][0]["fundingRate"])

funding["okx"] = {

"rate": rate,

"rate_pct": rate * 100,

"annualized": rate * 3 * 365 * 100,

}

except Exception as e:

funding["okx"] = {"error": str(e)}

# Average

rates = [d["rate"] for d in funding.values() if isinstance(d, dict) and "rate" in d]

if rates:

avg = sum(rates) / len(rates)

funding["average"] = {

"rate": avg,

"rate_pct": avg * 100,

"annualized": avg * 3 * 365 * 100,

}

funding["fetched_at"] = datetime.now(UTC).isoformat()

return funding

# -- long/short ratios --------------------------------------------------

def get_long_short_ratio(self) -> dict:

"""BTC long/short ratios from Binance (top traders + global)."""

ratios: dict = {}

# Top traders

try:

r = requests.get(

f"{self.binance_futures}/topLongShortPositionRatio"

f"?symbol=BTCUSDT&period=1h&limit=1",

timeout=5,

)

data = r.json()[0]

ratios["binance_top_traders"] = {

"long_ratio": float(data["longAccount"]),

"short_ratio": float(data["shortAccount"]),

"long_short_ratio": float(data["longShortRatio"]),

"timestamp": data["timestamp"],

}

except Exception as e:

ratios["binance_top_traders"] = {"error": str(e)}

# Global

try:

r = requests.get(

f"{self.binance_futures}/globalLongShortAccountRatio"

f"?symbol=BTCUSDT&period=1h&limit=1",

timeout=5,

)

data = r.json()[0]

ratios["binance_global"] = {

"long_ratio": float(data["longAccount"]),

"short_ratio": float(data["shortAccount"]),

"long_short_ratio": float(data["longShortRatio"]),

"timestamp": data["timestamp"],

}

except Exception as e:

ratios["binance_global"] = {"error": str(e)}

ratios["fetched_at"] = datetime.now(UTC).isoformat()

return ratios

# -- full snapshot ------------------------------------------------------

def get_full_snapshot(self, symbol: str = "BTC") -> dict:

"""Complete market snapshot: OI + funding + long/short."""

return {

"symbol": symbol,

"open_interest": self.get_btc_open_interest(),

"funding_rates": self.get_funding_rates(),

"long_short_ratio": self.get_long_short_ratio(),

"snapshot_time": datetime.now(UTC).isoformat(),

}

def format_snapshot(self, snapshot: dict) -> str:

"""Human-readable summary of a market snapshot."""

lines: list[str] = []

lines.append(f"--- {snapshot['symbol']} Market Data ---")

lines.append(f" {snapshot['snapshot_time'][:19]}\n")

# Open Interest

oi = snapshot["open_interest"]

lines.append("Open Interest:")

for ex in ["binance", "bybit", "okx"]:

if ex in oi and "oi_btc" in oi[ex]:

lines.append(f" {ex.title()}: {oi[ex]['oi_btc']:,.0f} BTC")

lines.append(f" Total: {oi.get('total_oi_btc', 0):,.0f} BTC\n")

# Funding Rates

fr = snapshot["funding_rates"]

lines.append("Funding Rates:")

for ex in ["binance", "bybit", "okx"]:

if ex in fr and "rate_pct" in fr[ex]:

rate = fr[ex]["rate_pct"]

lines.append(f" {ex.title()}: {rate:.4f}%")

if "average" in fr:

ann = fr["average"]["annualized"]

lines.append(f" Avg Annualized: {ann:.1f}%\n")

# Long/Short

ls = snapshot["long_short_ratio"]

if "binance_top_traders" in ls and "long_ratio" in ls["binance_top_traders"]:

top = ls["binance_top_traders"]

lines.append("Long/Short (Top Traders):")

lines.append(

f" Long: {top['long_ratio'] * 100:.1f}% | "

f"Short: {top['short_ratio'] * 100:.1f}%"

)

lines.append(f" Ratio: {top['long_short_ratio']:.2f}")

return "\n".join(lines)

# ---------------------------------------------------------------------------

# CLI

# ---------------------------------------------------------------------------

def main() -> None:

parser = argparse.ArgumentParser(description="Multi-exchange market data fetcher")

parser.add_argument("--symbol", default="BTC", help="Symbol (default: BTC)")

parser.add_argument(

"--json", action="store_true", help="Output raw JSON instead of formatted text"

)

args = parser.parse_args()

md = MarketData()

snapshot = md.get_full_snapshot(args.symbol)

if args.json:

print(json.dumps(snapshot, indent=2, default=str))

else:

print(md.format_snapshot(snapshot))

if __name__ == "__main__":

main()