在外汇 EA、汇率监控脚本、跨境资金分析程序开发中，欧元美元 fx_seurusd 是最常用的直盘品种。相比定时 HTTP 轮询，WebSocket 长连接可以做到价格变动瞬间推送，不存在轮询带来的延迟问题。本文完整覆盖连接创建、心跳维持、订阅指令发送、行情接收、数据解析、断线自动重连全流程，全部代码可直接复制本地运行，同时详细拆解服务端返回的完整 JSON 结构，方便业务逻辑开发。

import websocket
import json
import time
import threading

# 配置单一订阅品种
target_code = "fx_seurusd"

# 心跳循环函数，独立子线程运行
def send_heartbeat(ws_conn):
    while True:
        # 获取10位Unix时间戳
        timestamp = int(time.time())
        heartbeat_msg = json.dumps({"ping": timestamp})
        ws_conn.send(heartbeat_msg)
        time.sleep(10)

# 连接成功回调
def on_open(ws_conn):
    print("与行情服务建立长连接成功，准备订阅欧元美元fx_seurusd")
    # 组装订阅报文并发送
    sub_payload = json.dumps({"Key": target_code})
    ws_conn.send(sub_payload)
    # 启动心跳线程，守护主线程
    heartbeat_thread = threading.Thread(target=send_heartbeat, args=(ws_conn,), daemon=True)
    heartbeat_thread.start()

# 接收服务端推送的实时行情
def on_message(ws_conn, message_text):
    try:
        response_data = json.loads(message_text)
        # 判断是否为行情数据包
        if "body" in response_data:
            quote = response_data["body"]
            print("====== fx_seurusd 实时行情更新 ======")
            print(f"品种编码：{quote['StockCode']}")
            print(f"实时现价：{quote['Price']}")
            print(f"当日开盘价：{quote['Open']}")
            print(f"昨日收盘：{quote['LastClose']}")
            print(f"日内最高：{quote['High']}，日内最低：{quote['Low']}")
            print(f"买一价{quote['BP1']} 买一量{quote['BV1']}")
            print(f"卖一价{quote['SP1']} 卖一量{quote['SV1']}")
            print(f"单日涨跌额：{quote['Diff']}，涨跌幅：{quote['DiffRate']}%")
            print(f"更新时间戳：{quote['LastTime']}")
            print("=====================================\n")
    except Exception as err:
        print("行情数据解析异常：", str(err))

# 连接关闭回调，自动重启连接
def on_close(ws_conn, close_code, close_msg):
    print("WebSocket链路断开，3秒后执行自动重连逻辑")
    time.sleep(3)
    start_ws_client()

# 连接异常回调
def on_error(ws_conn, error_info):
    print("长连接通信异常：", str(error_info))

# 入口启动函数
def start_ws_client():
    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_client()


三、服务端返回完整 JSON 数据结构示例

Python WebSocket 完整流程订阅 fx_seurusd 欧元美元实时外汇行情

编程语言

详解欧元美元外汇实时行情WebSocket对接方案，提供完整Python代码实现，包含连接创建、心跳维持、数据订阅、行情解析和断线重连全流程。解析服务端返回的JSON数据结构，涵盖现价、开盘价、买卖盘口等关键字段，可直接用于外汇EA、汇率监控等金融系统开发。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Python WebSocket 完整流程订阅 fx_seurusd 欧元美元实时外汇行情-腾讯云开发者社区-腾讯云

Python WebSocket 完整流程订阅 fx_seurusd 欧元美元实时外汇行情

Python WebSocket 完整流程订阅 fx_seurusd 欧元美元实时外汇行情

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐