博客 / 詳情

返回

港美A股實時盤口數據：Python 接入示例教程

港美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://blog.itick.org/stock-api/hkus-stock-his-api-comparison-guide
GitHub：https://github.com/itick-org/

知識 , 後端 , Python
user avatar
0 位用戶收藏了這個故事！

發佈 評論

Some HTML is okay.