import asyncio import time import json import hashlib import websockets from websockets_proxy import Proxy, proxy_connect import gzip async def listen_ibit_futures(): uri = f"wss://wcws.hotcoins.cn/" # 代理服务器地址 proxy_url = "http://127.0.0.1:7899" proxy = Proxy.from_url(proxy_url) # 使用代理连接 async with proxy_connect(uri, proxy=proxy) as websocket: print(f"成功通过代理 {proxy_url} 连接到 {uri}") # 发送订阅消息 # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "tickers", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "deliver", "type": "tickers", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "order_fetch_kline", "granularity": "1min", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"type": "depth", "zip": False, "granularity": 100, "serialize": False}} # sub = {"event": "subscribe", "params": {"type": "tickers", "env": 0, "zip": False, "serialize": False}} sub = {"event": "subscribe", "params": {"type": "fund_rates", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "portal", "type": "orders", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "new_currency", "serialize": False, "env": 0}} # sub = {"event": "subscribe", "params": {"biz": "portal", "type": "condition_orders", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "portal", "type": "position", "granularity": "2mode", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "mark_candles", "zip": False, "contractCode": "btcusdt", "serialize": False, "env": 0}} # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "candles", "zip": False, "granularity": "15min", "serialize": False, "env": 0}} sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "fund_rate", "zip": False, "contractCode": "btcusdt", "serialize": False, "env": 0}} # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "ticker", "zip": False, "contractCode": "btcusdt", "serialize": False, "env": 0}} # sub = {"event": "subscribe", # "params": {"biz": "perpetual", "type": "order_fetch_kline", "contractCode": "btcusdt", "granularity": "1min", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "subscribe", "params": {"biz": "perpetual", "type": "depth", "contractCode": "btcusdt", "zip": False, "granularity": 100, "serialize": False, "env": 0}} # sub = {"event": "unsubscribe", "params": {"biz": "perpetual", "type": "order_fetch_kline", "granularity": "1min", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "unsubscribe", "params": {"type": "tickers", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "unsubscribe", "params": {"type": "fund_rates", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "unsubscribe", "params": {"type": "fund_rates", "env": 0, "zip": False, "serialize": False}} # sub = {"event": "unsubscribe", "params": {"type": "depth", "zip": False, "granularity": 100, "serialize": False}} # sub = {"event": "subscribe", # "params": {"biz": "perpetual", "type": "candles", "zip": False, "contractCode": "btcusdt", "granularity": "15min", "serialize": False, "env": 0}} await websocket.send(json.dumps(sub)) print("已发送订阅消息") # 定义异步心跳函数 async def send_heartbeat(): while True: heartbeat_message = {"event": "ping"} await websocket.send(json.dumps(heartbeat_message)) print(f"发送心跳消息: {heartbeat_message}") await asyncio.sleep(30) # 启动异步心跳任务 heartbeat_task = asyncio.create_task(send_heartbeat()) try: # 接收消息循环 while True: message = await websocket.recv() # 处理二进制数据（gzip压缩） if isinstance(message, bytes): # 检查是否是gzip压缩数据 if message.startswith(b'\x1f\x8b'): try: # 解压gzip数据 decompressed_data = gzip.decompress(message) message = decompressed_data.decode('utf-8') except Exception as e: print(f"解压数据失败: {e}") print(f"原始数据 (前50字节): {message[:50]}") continue else: # 尝试多种编码解码非gzip二进制数据 decoded = False # 尝试UTF-8解码 try: message = message.decode('utf-8') decoded = True except UnicodeDecodeError: pass # 如果UTF-8解码失败，尝试其他编码或以十六进制形式显示 if not decoded: try: message = message.decode('gbk') decoded = True except UnicodeDecodeError: pass if not decoded: # 如果所有解码都失败，以十六进制形式显示数据 print(f"无法解码的二进制数据 (前100字节): {message[:100].hex()}") continue print(f"接收到消息: {message}") # 解析JSON数据 try: data = json.loads(message) except json.JSONDecodeError as e: print(f"JSON解析失败: {e}") print(f"原始消息: {message}") continue if 'channel' in data and data['channel'] == 'pong': continue # 根据不同频道处理数据 if 'channel' in data: channel = data['channel'] if channel == 'account' and 'data' in data: # 账户信息 for account in data['data']: print(f"账户余额: {account.get('balance', 'N/A')} {account.get('currency', 'N/A')}") elif channel == 'currentPositions' and 'data' in data: # 当前持仓 for position in data['data']: print(f"持仓: {position}") elif channel in ['openOrders', 'planCloseOrders', 'planOpenOrders'] and 'data' in data: # 订单信息 for order in data['data']: # 安全访问 'p' 字段，如果不存在则显示 N/A price = order.get('p', 'N/A') print(f"订单价格: {price}") # 其他频道可以按需添加处理逻辑 except websockets.exceptions.ConnectionClosed as e: print(f"连接关闭: {e}") except Exception as e: print(f"发生错误: {e}") import traceback traceback.print_exc() finally: # 取消心跳任务 heartbeat_task.cancel() try: await heartbeat_task except asyncio.CancelledError: pass async def main(): await listen_ibit_futures() if __name__ == "__main__": asyncio.run(main())