#!/usr/bin/env python # -*- coding: utf-8 -*- """ 交易所清算数据统一接口 数据源: 1. OKX WebSocket (liquidation-orders) — 实时清算事件 2. OKX REST API — 历史清算单（仅MARGIN） 3. Coinglass API — 行业清算数据汇总 """ import sys, os, json, time, requests from collections import deque from datetime import datetime, timedelta from typing import Optional, Dict, List sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) import config CG_URL = "https://open-api.coinglass.com/api/pro/v1" CG_HEADERS = {"apiKey": config.CG_API_KEY, "Accept": "application/json"} class LiquidationSource: """清算数据源枚举""" OKX_WS = "okx_ws" OKX_REST = "okx_rest" COINGLASS = "coinglass" class LiquidationEvent: """统一清算事件格式""" def __init__(self, source: str, symbol: str, side: str, size: float, price: float, ts: int, raw: dict = None): self.source = source # 数据源 self.symbol = symbol # 合约名 LAB-USDT-SWAP self.side = side # 'buy'=多头被清算（看跌）/'sell'=空头被清算（看涨） self.size = size # 张数 self.price = price # 清算价格 self.ts = ts # 时间戳(ms) self.value = size * price # 清算金额等值（美元） self.raw = raw # 原始数据 def __repr__(self): return (f"[{self.source}] {self.symbol} {self.side:>4s} " f"{self.size:.2f}张 @ ${self.price:.4f} ({self.value:.0f}USD)") class CoinGlassAPI: """Coinglass 清算数据查询""" @staticmethod def liquidation_map(limit: int = 10) -> List[dict]: """清算热力图 - 各币种清算总金额""" try: r = requests.get(f"{CG_URL}/futures/liquidation_map", params={"limit": limit, "ex": "all"}, headers=CG_HEADERS, proxies=config.PROXIES, timeout=10) d = r.json() if d.get("code") == "0": return d.get("data", []) print(f"Coinglass liq_map: {d}") return [] except Exception as e: print(f"Coinglass liq_map error: {e}") return [] @staticmethod def liquidation_by_symbol(symbol: str, interval: str = "1h", limit: int = 24) -> List[dict]: """单币种清算历史""" try: r = requests.get(f"{CG_URL}/futures/liquidation_chart", params={"symbol": symbol.replace("-", "").replace("USDT", "USDT"), "interval": interval, "limit": limit}, headers=CG_HEADERS, proxies=config.PROXIES, timeout=10) d = r.json() if d.get("code") == "0": return d.get("data", []) return [] except Exception as e: print(f"Coinglass liq_chart {symbol}: {e}") return [] @staticmethod def liquidation_order_list(symbol: str = "ALL", limit: int = 20) -> List[dict]: """逐笔清算订单""" try: r = requests.get(f"{CG_URL}/futures/liquidation_order_list", params={"symbol": symbol, "limit": limit}, headers=CG_HEADERS, proxies=config.PROXIES, timeout=10) d = r.json() if d.get("code") == "0": return d.get("data", []) return [] except Exception as e: print(f"Coinglass liq_orders {symbol}: {e}") return [] class OKXLiquidationAPI: """OKX REST清算数据""" @staticmethod def get_liquidation_orders(inst_type: str = "MARGIN", inst_id: str = None, state: str = "filled", limit: int = 10) -> List[LiquidationEvent]: """ 清算订单历史（仅MARGIN支持，SWAP需用WS） """ params = {"instType": inst_type, "state": state, "limit": limit} if inst_id: params["instId"] = inst_id try: r = requests.get(f"{config.OKX_BASE_URL}/api/v5/public/liquidation-orders", params=params, proxies=config.PROXIES, timeout=10) data = r.json() if data.get("code") != "0": print(f"OKX liq REST: {data}") return [] events = [] for d in data.get("data", []): events.append(LiquidationEvent( source="okx_rest", symbol=d.get("instId", ""), side=d.get("side", ""), size=float(d.get("sz", 0)), price=float(d.get("fillPx", 0)), ts=int(d.get("ts", 0)), raw=d )) return events except Exception as e: print(f"OKX liq REST error: {e}") return [] class LiquidationTracker: """ 统一清算追踪器 - 接收 WS 实时事件 - 定时轮询 REST/Coinglass 补充 - 提供聚合指标 (I_liq, 清算密度等) """ def __init__(self, window_sec: int = 5): self.events = deque(maxlen=500) # 全部事件 self.window_ns = window_sec * 1_000_000_000 self.last_cg_poll = 0 def add_event(self, evt: LiquidationEvent): """添加一条清算事件""" self.events.append(evt) self._prune() def add_ws_raw(self, symbol: str, side: str, sz: float, price: float, ts: int, raw: dict = None): """从WS原始数据添加事件""" evt = LiquidationEvent("okx_ws", symbol, side, sz, price, ts, raw) self.add_event(evt) def _prune(self): """清理过期事件""" cutoff = time.time_ns() - self.window_ns * 2 while self.events and self.events[0].ts * 1_000_000 < cutoff: self.events.popleft() def recent_events(self, window_sec: float = 3.0): """最近N秒内的事件""" cutoff = (time.time() - window_sec) * 1_000_000_000 return [e for e in self.events if e.ts * 1_000_000 > cutoff] def I_liq(self, window_sec: float = 3.0, mid_price: float = None) -> float: """清算强度（美元总额/秒）""" events = self.recent_events(window_sec) if not events: return 0.0 total_value = sum(e.value for e in events) return total_value / window_sec def liq_density(self, window_sec: float = 3.0) -> float: """清算密集度（事件数/秒）""" events = self.recent_events(window_sec) return len(events) / max(window_sec, 0.1) def liq_net_direction(self, window_sec: float = 3.0) -> float: """[-1, 1] 正值=多头清算更密集（看跌）""" events = self.recent_events(window_sec) if not events: return 0.0 buy_val = sum(e.value for e in events if e.side == "buy") # 多头→看跌 sell_val = sum(e.value for e in events if e.side == "sell") # 空头→看涨 total = buy_val + sell_val if total == 0: return 0.0 # buy=多头清算多→看空→负值 return (sell_val - buy_val) / total def summary(self, window_sec: float = 3.0) -> dict: """清算状态摘要""" events = self.recent_events(window_sec) return { "total_events": len(events), "I_liq": self.I_liq(window_sec), "density": self.liq_density(window_sec), "net_dir": self.liq_net_direction(window_sec), "buy_vol": sum(e.value for e in events if e.side == "buy"), "sell_vol": sum(e.value for e in events if e.side == "sell"), } def poll_coinglass(self, symbol: str = "LABUSDT"): """轮询Coinglass补充数据（每分钟一次）""" now = time.time() if now - self.last_cg_poll < 60: return self.last_cg_poll = now orders = CoinGlassAPI.liquidation_order_list(symbol, limit=10) if orders: for o in orders: self.add_event(LiquidationEvent( source="coinglass", symbol=o.get("symbol", symbol), side='buy' if float(o.get('longLiquidation', 0)) > 0 else 'sell', size=float(o.get('longLiquidation', o.get('shortLiquidation', 0))), price=float(o.get('price', 0)), ts=int(o.get('updateTime', 0) / 1000), raw=o )) def poll_okx_rest(self, inst_id: str = None): """轮询OKX REST补充（每30秒一次）""" try: events = OKXLiquidationAPI.get_liquidation_orders( inst_type="MARGIN", inst_id=inst_id, limit=5) for e in events: self.events.append(e) except: pass # ── 快速测试 ── if __name__ == "__main__": tracker = LiquidationTracker() print("=== OKX REST 清算 (MARGIN) ===") events = OKXLiquidationAPI.get_liquidation_orders(limit=3) for e in events: print(f" {e}") tracker.add_event(e) print("\n=== Coinglass 清算热力图 (Top 5) ===") lmap = CoinGlassAPI.liquidation_map(limit=5) for item in lmap: liq = item.get('longLiquidation', 0) or item.get('shortLiquidation', 0) print(f" {item.get('symbol','?'):10s} | 多头清={item.get('longLiquidation',0):>12.0f} " f"| 空头清={item.get('shortLiquidation',0):>12.0f}") print(f"\n=== 追踪器摘要 ===") print(json.dumps(tracker.summary(), indent=2))