mobile-rpa/testcase/main-mexc.py

import asyncio
import websockets
import json
# 1. 从扩展库导入代理连接函数和代理类
from websockets_proxy import Proxy, proxy_connect


async def listen_mexc_futures():
    uri = "wss://futures.mexc.com/edge"

    # 2. 定义你的代理服务器地址和端口
    # 请替换成你自己的代理服务器信息
    # 支持 http, socks4, socks5
    proxy_url = "http://127.0.0.1:7899"  # 这是一个HTTP代理的例子

    proxy = Proxy.from_url(proxy_url)

    # 3. 使用 proxy_connect 替代 websockets.connect
    async with proxy_connect(uri, proxy=proxy) as websocket:
        print(f"成功通过代理 {proxy_url} 连接到 {uri}")

        # 定义异步心跳函数
        async def send_subscription():

            # 取消订阅增量盘口列表 (通常在开始新的订阅前，先取消旧的)
            unsub_tick_list_inc = {"method": "unsub.tick.list.inc", "param": {"timezone": "UTC+8"}}
            # await websocket.send(json.dumps(unsub_tick_list_inc))

            # 取消订阅全量盘口列表
            unsub_tick_list = {"method": "unsub.tick.list", "param": {"timezone": "UTC+8"}}
            # await websocket.send(json.dumps(unsub_tick_list))

            # 订阅所有合约的基础信息 (例如合约名称、精度等)
            sub_contract = {"method": "sub.contract"}
            # await websocket.send(json.dumps(sub_contract))

            # 订阅个人用户偏好设置
            sub_personal_user_preference = {"method": "sub.personal.user.preference"}
            # await websocket.send(json.dumps(sub_personal_user_preference))

            # 批量订阅多个交易对的最新价(Ticker)信息
            sub_tick_batch = {"method": "sub.tick.batch", "param": {"timezone": "UTC+8",
                                                                    "symbols": ["DOGE_USD", "ETH_USDT", "BR_USDT", "SOL_USDT", "PI_USDT", "XRP_USDT", "ORCA_USDT", "TUT_USDT",
                                                                                "MUBARAK_USDT", "BUBB_USDT", "SIREN_USDT", "MUBARAKAH_USDT", "TAT_USDT", "BANANAS31_USDT",
                                                                                "SZN_USDT", "YZYSOL_USDT", "DLC_USDT", "FORM_USDT", "SLING_USDT", "NIL_USDT"]}}
            # await websocket.send(json.dumps(sub_tick_batch))

            # 订阅单个交易对的深度盘口数据 (按价格步长0.1聚合)
            sub_depth_step = {"method": "sub.depth.step", "param": {"symbol": "DOGE_USD", "step": 0.1}}
            # await websocket.send(json.dumps(sub_depth_step))

            # 订阅单个交易对的实时成交记录
            sub_deal = {"method": "sub.deal", "param": {"symbol": "DOGE_USD", "compress": "true"}}
            # await websocket.send(json.dumps(sub_deal))

            # 订阅单个交易对的资金费率
            sub_funding_rate = {"method": "sub.funding.rate", "param": {"symbol": "DOGE_USD"}}
            # await websocket.send(json.dumps(sub_funding_rate))

            # 订阅单个交易对的合理价格 (也叫标记价格，用于计算未实现盈亏)
            sub_fair_price = {"method": "sub.fair.price", "param": {"symbol": "DOGE_USD"}}
            await websocket.send(json.dumps(sub_fair_price))

            # 订阅单个交易对的指数价格 (通常是多个主流交易所现货价格的加权平均)
            sub_index_price = {"method": "sub.index.price", "param": {"symbol": "DOGE_USD"}}
            # await websocket.send(json.dumps(sub_index_price))

            # 订阅单个交易对的K线数据 (这里订阅的是15分钟K线)
            sub_kline = {"method": "sub.kline", "param": {"symbol": "DOGE_USD", "interval": "Min15"}}
            # await websocket.send(json.dumps(sub_kline))

            while True:
                heartbeat_message = {"method": "ping"}
                await websocket.send(json.dumps(heartbeat_message))
                print(f"发送心跳消息: {heartbeat_message}")
                await asyncio.sleep(30)

        # 启动异步心跳任务
        asyncio.create_task(send_subscription())
        while True:
            try:
                message = await websocket.recv()
                print(f"接收到消息: {message}")
                data = json.loads(message)

                if 'channel' in data and data['channel'] == 'pong':
                    continue

                if 'data' in data and isinstance(data['data'], list) and len(data['data']) > 0:
                    for deal in data['data']:
                        print(f"成交价格: {deal['p']}")

            except websockets.exceptions.ConnectionClosed as e:
                print(f"连接关闭: {e}")
                break
            except Exception as e:
                print(f"发生错误: {e}")
                break


async def main():
    await listen_mexc_futures()


if __name__ == "__main__":
    asyncio.run(main())