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

{
  "code": 0,
  "data": {
    "700": {
      "s": "700",
      "a": [
        /* 卖盘数组 */
      ],
      "b": [
        /* 买盘数组 */
      ]
    },
    "9988": {
      /* 类似 */
    }
  }
}

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)

{
  "code": 1,
  "data": {
    "s": "AAPL.US",
    "a": [
      /* 卖盘 */
    ],
    "b": [
      /* 买盘 */
    ],
    "type": "depth"
  }
}

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()