港美A股实时盘口数据：Python 接入示例教程

在量化交易和大宗交易决策中，实时盘口数据（也称为市场深度或 Level 2 数据）是关键要素。它展示了股票买卖盘的挂单价格、数量和订单数，帮助投资者洞察市场供需动态、预测价格走势，并优化交易策略。本文聚焦港股（HK）、美股（US）和 A 股（SH/SZ）的实时盘口，介绍如何通过 RESTful API 和 WebSocket 方式接入这些数据，支持多市场覆盖和毫秒级更新。

作为开发者，你可以通过 Python 代码快速集成类似 API。本文提供完整示例，适用于教育和学习目的。注意：实际使用需替换为你的 API 配置，并遵守相关平台规则。

什么是实时盘口？

实时盘口指的是股票买卖双方的挂单信息，通常包括多档（例如 5-10 档）买盘（Bid）和卖盘（Ask）。每个档位包含：

• 价格（Price）：挂单价格。

• 挂单量（Volume）：该价格下的总挂单数量。

• 订单数量（Orders）：该价格下的订单数。

例如，在港股市场，腾讯控股（700.HK）的盘口可能显示买一价 568 HKD，挂单量 75000 股。这类数据对高频交易者至关重要，因为它揭示了市场的“深度”——浅盘口可能意味着流动性不足，容易出现价格波动。

支持的地区包括：

• 港股 (HK)：如 700 (腾讯)、9988 (阿里巴巴)。

• 美股 (US)：如 AAPL (苹果)、TSLA (特斯拉)。

• A 股 (SH/SZ)：上海和深圳交易所股票。

接入方式有批量查询（REST）和实时推送（WebSocket）。

批量实时盘口 API（RESTful）

如果你需要一次性获取多个股票的盘口数据，推荐使用 GET /stock/depths 接口。它支持批量 codes 查询，最多 10 档盘口，返回 JSON 格式。

请求参数

• region：市场代码，必填（如 HK、US、SH、SZ）。

• codes：股票代码，逗号分隔（如 700,9988）。

响应示例

响应中，”a” 表示卖盘（Ask），”b” 表示买盘（Bid）。以下是 HK 地区 700 和 9988 的示例响应（简化版）：

{

 "code": 0,

 "data": {

 "700": {

 "s": "700",

 "a": [

 /* 卖盘数组 */

      ],

 "b": [

 /* 买盘数组 */

      ]

    },

 "9988": {

 /* 类似 */

    }

  }

}

Python 接入示例

使用 requests 库发送 GET 请求。确保在 headers 中添加你的 Token。

import requests

# API 基础 URL 和你的 Token

BASE_URL = "https://api.itick.org/stock/depths"

API_TOKEN = "your_token" # 替换为你的真实 Token

# 请求参数

params = {

 "region": "HK",  # 可替换为 US, SH, SZ 等

 "codes": "700,9988" # 支持批量，逗号分隔

}

headers = {

 "accept": "application/json",

 "token": API_TOKEN

}

# 发送请求

response = requests.get(BASE_URL, params=params, headers=headers)

# 处理响应

if response.status_code == 200:

    data = response.json()

 if data["code"] == 0:

 print("成功获取盘口数据：")

 for stock, details in data["data"].items():

 print(f"股票 {stock}:")

 print("卖盘 (Ask):")

 for ask in details["a"]:

 print(f"档位 {ask['po']}: 价格 {ask['p']}, 挂单量 {ask['v']}, 订单数 {ask['o']}")

 print("买盘 (Bid):")

 for bid in details["b"]:

 print(f"档位 {bid['po']}: 价格 {bid['p']}, 挂单量 {bid['v']}, 订单数 {bid['o']}")

 else:

 print("API 错误:", data["msg"])

else:

 print("HTTP 错误:", response.status_code)

这个示例适用于批量查询港美 A 股的盘口。如果你需要 A 股数据，将 region 改为 “SH” 或 “SZ”，codes 替换为相应股票代码（如 600000）。

实时盘口推送（WebSocket）

对于需要毫秒级更新的场景，WebSocket 是首选。它支持订阅多个股票的 depth（盘口）、quote（报价）和 tick（成交）数据。连接后，服务器会推送更新。

接入流程

1. 连接：使用 WebSocket 地址，带授权头。

2. 验证：自动验证，成功后收到 “authenticated”。

3. 订阅：发送订阅消息，指定 params（股票，如 AAPL$US）和 types（depth 等）。

4. 心跳：每 30 秒发送 ping 保持连接。

5. 响应：实时接收盘口更新。

响应示例

盘口推送（type: “depth”）：

{

 "code": 1,

 "data": {

 "s": "AAPL.US",

 "a": [

 /* 卖盘 */

    ],

 "b": [

 /* 买盘 */

    ],

 "type": "depth"

  }

}

Python 接入示例

使用 websocket 库实现。代码包括连接、订阅、心跳和消息处理。支持港美 A 股订阅（如 params: "700$HK,AAPL$US,600000$SH“）。

import websocket

import json

import threading

import time

# WebSocket URL 和你的 Token

WS_URL = "wss://api.itick.org/stock"

API_TOKEN = "your_token" # 替换为你的真实 Token

def  on_message(ws, message):

 """处理接收到的消息"""

 print("Received message:", message)

    data = json.loads(message)

 # 处理连接和认证

 if data.get("msg") == "Connected Successfully":

 print("Connected successfully")

 elif data.get("resAc") == "auth":

 if data.get("code") == 1:

 print("Authentication successful")

            subscribe(ws)

 else:

 print("Authentication failed")

            ws.close()

 elif data.get("resAc") == "subscribe":

 if data.get("code") == 1:

 print("Subscription successful")

 else:

 print("Subscription failed:", data.get("msg"))

 elif data.get("data"):

        market_data = data["data"]

        data_type = market_data.get("type")

        symbol = market_data.get("s")

 if data_type == "depth":

 print(f"实时盘口更新 for {symbol}:")

 print("卖盘 (Ask):", market_data["a"])

 print("买盘 (Bid):", market_data["b"])

def  on_error(ws, error):

 print("Error:", error)

def  on_close(ws, close_status_code, close_msg):

 print("Connection closed")

def  on_open(ws):

 print("WebSocket connection opened")

def  subscribe(ws):

 """订阅港美A股盘口"""

    subscribe_msg = {

 "ac": "subscribe",

 "params": "700$HK,AAPL$US,600000$SH",  # 支持多市场批量订阅

 "types": "depth" # 只订阅盘口，也可加 quote,tick

    }

    ws.send(json.dumps(subscribe_msg))

 print("Subscribe message sent")

def  send_ping(ws):

 """心跳机制"""

 while  True:

        time.sleep(30)

        ping_msg = {

 "ac": "ping",

 "params": str(int(time.time() * 1000))

        }

        ws.send(json.dumps(ping_msg))

 print("Ping sent")

if  __name__ == "__main__":

 # 创建 WebSocket 连接

    ws = websocket.WebSocketApp(

        WS_URL,

 header={"token": API_TOKEN},

 on_open=on_open,

 on_message=on_message,

 on_error=on_error,

 on_close=on_close

    )

 # 启动心跳线程

    ping_thread = threading.Thread(target=send_ping, args=(ws,))

ping_thread.daemon = True

    ping_thread.start()

 # 运行 WebSocket

    ws.run_forever()

这个示例会实时打印盘口更新。适用于监控多市场数据。

注意事项与优化建议

• 限额与授权：确保你的配置支持订阅上限。

• 多市场支持：params 使用 “code$region“ 格式，确保正确（如 700$HK）。

• 错误处理：在生产环境中，添加重连机制和日志记录。

• 应用场景：结合 Pandas 处理数据，或集成到量化平台。

• 安全：令牌勿泄露，使用环境变量存储。

总结

港美 A 实时盘口数据是交易决策的重要基础，通过高效的 API 获取多市场、多股票的深度数据。本文提供的多语言实现示例可直接作为开发模板，结合具体业务场景解析数据，能快速搭建盘口监控或交易策略系统。在实际应用中，需注意数据实时性与稳定性的平衡，充分发挥盘口数据的价值。

温馨提示：本文仅供参考，不构成任何投资建议。市场有风险，投资需谨慎

参考文档：https://docs.itick.org/

GitHub：https://github.com/itick-org/

本作品采用《CC 协议》，转载必须注明作者和本文链接