Python WebSocket 实时订阅 usdcnh 离岸人民币汇率完整开发流程

火柴oo

239人浏览 · 2026-07-09 15:12:43

火柴oo · 2026-07-09 15:12:43 发布

在跨境外贸结算、离岸资产对冲、人民币汇率趋势分析程序开发场景中，美元离岸人民币 usdcnh 是核心监测货币对。轮询方式会产生固定延迟，采用 WebSocket 长连接能够实时捕捉汇率瞬间波动，完整覆盖心跳保活、断线重连、数据包解析全流程，下文全部代码可直接本地运行，附带完整服务端返回 JSON 结构，便于业务逻辑开发。

一、通信基础规范

1、长连接统一地址：ws://39.107.99.235/ws

2、目标订阅编码：usdcnh

3、心跳强制规则：客户端每 10 秒推送 {"ping": 十位 Unix 时间戳}，长时间无心跳会被服务端断开连接

4、订阅报文格式：连接成功后发送 {"Key":"品种编码"}，单品种直接填写 usdcnh

5、推送机制：仅价格、盘口产生变动时下发数据包，无重复冗余推送

二、完整可运行 Python 代码

import websocket
import json
import time
import threading

# 配置订阅标的
target_code = "fx_susdcnh"

def heartbeat_task(ws_conn):
    # 独立子线程持续发送心跳
    while True:
        timestamp = int(time.time())
        ws_conn.send(json.dumps({"ping": timestamp}))
        time.sleep(10)

def on_open(ws_conn):
    print("离岸人民币行情长连接建立完成，发起订阅")
    sub_msg = json.dumps({"Key": target_code})
    ws_conn.send(sub_msg)
    # 启动守护心跳线程
    threading.Thread(target=heartbeat_task, args=(ws_conn,), daemon=True).start()

def on_message(ws_conn, msg_text):
    try:
        raw_data = json.loads(msg_text)
        if "body" in raw_data:
            quote_info = raw_data["body"]
            print("==== fx_susdcnh 离岸人民币实时行情 ====")
            print(f"品种编码：{quote_info['StockCode']}")
            print(f"实时汇率：{quote_info['Price']}")
            print(f"当日开盘：{quote_info['Open']}")
            print(f"昨日收盘：{quote_info['LastClose']}")
            print(f"日内高低区间：{quote_info['High']}/{quote_info['Low']}")
            print(f"买一档报价：{quote_info['BP1']}，卖一档报价：{quote_info['SP1']}")
            print(f"单日涨跌额：{quote_info['Diff']}，涨跌幅：{quote_info['DiffRate']}%")
            print(f"数据更新时间戳：{quote_info['LastTime']}")
            print("========================================\n")
    except Exception as e:
        print("行情数据包解析异常：", str(e))

def on_close(ws_conn, close_code, tips):
    print("离岸人民币行情连接断开，3秒后自动重连")
    time.sleep(3)
    start_ws_service()

def on_error(ws_conn, err_info):
    print("长连接通信异常：", str(err_info))

def start_ws_service():
    ws_app = websocket.WebSocketApp(
        url="ws://39.107.99.235/ws",
        on_open=on_open,
        on_message=on_message,
        on_close=on_close,
        on_error=on_error
    )
    ws_app.run_forever()

if __name__ == "__main__":
    start_ws_service()

三、服务端完整返回 JSON 样例

{
    "body": {
        "StockCode": "fx_susdcnh",
        "Price": 7.2468,
        "Open": 7.2312,
        "LastClose": 7.2295,
        "High": 7.2513,
        "Low": 7.2241,
        "Time": "2026-07-09 10:18:32",
        "LastTime": 1783664312,
        "BP1": 7.2462,
        "BV1": 3210,
        "SP1": 7.2468,
        "SV1": 2876,
        "TotalVol": "",
        "Diff": 0.0173,
        "DiffRate": 0.24
    }
}