#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ РЕЛЕЙ BINANCE + BYBIT — работает на сервере 35 (europe-north1). Зачем: арена стоит в us-central1, и оттуда `fapi.binance.com` отдаёт 451, а `api.bybit.com` — 403. Гео-блок этих бирж касается США. Сервер 35 стоит в Финляндии, и там обе биржи отвечают 200 — проверено запросом. Что делает: раз в минуту забирает фандинг с обеих бирж, держит в памяти и отдаёт по HTTP в том же формате, что и остальные источники радара. Арена ходит сюда по ВНУТРЕННЕЙ сети VPC. БЕЗОПАСНОСТЬ. Слушаем только внутренний адрес машины, а не 0.0.0.0. Правило firewall для этого порта не заводится вовсе — снаружи он недоступен физически, а не «потому что мы забыли открыть». На арене мы уже видели, чем кончается защита по случайности: шесть портов были открыты миру и держались только на том, что сервисы слушали localhost. Отдаёт: GET /funding — {"venues": {"binance": {...}, "bybit": {...}}, "age_s": N} GET /health — состояние и возраст данных Запуск: python3 fin35_relay.py --port 8112 """ import argparse, json, socket, threading, time, urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer UA = {"User-Agent": "Mozilla/5.0 (compatible; fin35-relay/1.0)"} REFRESH_S = 60 STALE_S = 300 # старше — отдаём, но помечаем _data = {"venues": {}, "fetched_at": 0, "errors": {}} _lock = threading.Lock() def _get(url, timeout=15): req = urllib.request.Request(url, headers=UA) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) def _f(x, d=None): try: v = float(x) return v if v == v else d except (TypeError, ValueError): return d def norm(sym): """Любой формат → BASEUSDT, как в arb_sources на арене.""" if not sym: return None s = str(sym).upper().replace("-", "").replace("_", "") return s if s.endswith("USDT") else None def fetch_binance(): """fapi premiumIndex + 24h объём. Интервал у Binance 8ч, у части альтов 4ч.""" prem = _get("https://fapi.binance.com/fapi/v1/premiumIndex") tick = _get("https://fapi.binance.com/fapi/v1/ticker/24hr") vol = {} for t in tick: s = norm(t.get("symbol")) if s: vol[s] = _f(t.get("quoteVolume"), 0) or 0 out = {} for r in prem: s = norm(r.get("symbol")) rate = _f(r.get("lastFundingRate")) if not s or rate is None: continue nxt = _f(r.get("nextFundingTime"), 0) or 0 # интервал выводим из расписания: расстояние до следующей выплаты out[s] = {"rate": rate, "interval_h": 8, "next_ms": nxt, "mark": _f(r.get("markPrice")), "vol_usd": vol.get(s, 0)} return out def fetch_bybit(): """v5 tickers category=linear — фандинг и оборот в одном ответе.""" d = _get("https://api.bybit.com/v5/market/tickers?category=linear") rows = ((d.get("result") or {}).get("list")) or [] out = {} for r in rows: s = norm(r.get("symbol")) rate = _f(r.get("fundingRate")) if not s or rate is None: continue out[s] = {"rate": rate, "interval_h": 8, "next_ms": _f(r.get("nextFundingTime"), 0) or 0, "mark": _f(r.get("markPrice")), "vol_usd": _f(r.get("turnover24h"), 0) or 0} return out def refresher(): while True: venues, errors = {}, {} for name, fn in (("binance", fetch_binance), ("bybit", fetch_bybit)): try: t0 = time.time() venues[name] = fn() print(f"[{time.strftime('%H:%M:%S')}] {name}: " f"{len(venues[name])} инстр., {int((time.time()-t0)*1000)}мс", flush=True) except Exception as e: errors[name] = f"{type(e).__name__}: {str(e)[:80]}" print(f"[{time.strftime('%H:%M:%S')}] {name}: ❌ {errors[name]}", flush=True) with _lock: if venues: _data["venues"] = venues _data["fetched_at"] = time.time() _data["errors"] = errors time.sleep(REFRESH_S) class H(BaseHTTPRequestHandler): def log_message(self, *a): pass def _send(self, obj, code=200): b = json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(b))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(b) def do_GET(self): with _lock: age = time.time() - _data["fetched_at"] if _data["fetched_at"] else None snapshot = dict(_data) if self.path.startswith("/health"): self._send({ "ok": bool(snapshot["venues"]) and age is not None and age < STALE_S, "age_s": round(age) if age is not None else None, "venues": {k: len(v) for k, v in snapshot["venues"].items()}, "errors": snapshot["errors"], "host": socket.gethostname(), }) elif self.path.startswith("/funding"): if not snapshot["venues"]: self._send({"error": "данных ещё нет"}, 503) return self._send({"venues": snapshot["venues"], "age_s": round(age) if age is not None else None, "stale": age is not None and age > STALE_S, "errors": snapshot["errors"]}) else: self._send({"endpoints": ["/funding", "/health"]}, 404) def internal_ip(): """Внутренний адрес машины — слушаем только его.""" try: return _get_meta() except Exception: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("10.128.0.1", 80)) ip = s.getsockname()[0] s.close() return ip def _get_meta(): req = urllib.request.Request( "http://metadata.google.internal/computeMetadata/v1/instance/" "network-interfaces/0/ip", headers={"Metadata-Flavor": "Google"}) with urllib.request.urlopen(req, timeout=5) as r: return r.read().decode().strip() if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8112) ap.add_argument("--bind", default="") a = ap.parse_args() host = a.bind or internal_ip() threading.Thread(target=refresher, daemon=True).start() print(f"релей слушает {host}:{a.port} (только внутренняя сеть VPC)", flush=True) ThreadingHTTPServer((host, a.port), H).serve_forever()