港股实时行情 API 接口接入实战指南：从选型到落地全解析

在量化交易和金融科技应用开发中，实时行情数据是决策的核心。对于港股市场而言，选择合适的数据接口并正确接入，是构建稳定交易系统的第一步。

随着港股市场与 A 股市场的互联互通机制不断深化，港股实时行情数据的需求日益增长。无论是构建量化交易系统、开发投资分析工具，还是打造金融资讯平台，都离不开稳定可靠的行情数据接口。

本文将全面解析港股实时行情 API 接口的选型与接入过程，提供从主流接口对比、Python 实战代码到生产环境部署的完整指南。

一、港股行情 API 接口选型指南

目前市场上港股行情 API 鱼龙混杂，从传统金融数据服务商到新兴技术平台，各有优劣。我们从开发者最关注的实时性、成本、易用性、数据完整性四个维度，对比 itick 与两款主流接口的核心差异：

二、接入前置准备：统一 API 请求头设置

无论选择哪个 API 服务商，合理的请求头设置是确保 API 调用成功的基础。以下是根据行业标准整理的统一请求头配置：

获取 token，请参考 官网文档

import requests

import os

from typing import Dict, Any

# 统一请求头配置

headers = {

 "accept": "application/json",

 "token": os.getenv("HK_STOCK_API_TOKEN", "yourtoken"),  # 从环境变量读取token，增强安全性

 "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",

 "content-type": "application/json; charset=utf-8"

}

class  HKStockAPI:

 def  __init__(self, base_url="https://api.itick.org", custom_headers=None):

 self.base_url = base_url

 self.headers = headers.copy()

 if custom_headers:

 self.headers.update(custom_headers)

 def  _make_request(self, endpoint, params=None):

 """统一封装的请求方法"""

 try:

url = f"{self.base_url}{endpoint}"

            response = requests.get(

                url,

 headers=self.headers,

 params=params,

 timeout=10

            )

            response.raise_for_status()  # 检查HTTP状态码

 return response.json()

 except requests.exceptions.RequestException as e:

 print(f"API请求失败: {e}")

 return  None

三、港股实时行情 API 接口实战（代码示例）

1. 股票实时报价接口

实时报价是行情数据中最基础也是最重要的接口，提供最新的成交价、涨跌幅、成交量等核心数据。

def  get_stock_quote(self, symbol, region="HK"):

 """

    获取港股实时报价

    :param symbol: 股票代码，如 '700' 对应腾讯控股

    :param region: 地区代码，港股固定为 'HK'

    :return: 实时报价数据字典

    """

endpoint = "/stock/quote"

    params = {

 "region": region,

 "code": symbol

    }

data = self._make_request(endpoint, params)

 if data and data.get("code") == 0:

 return data.get("data")

 else:

 print(f"获取实时报价失败: {data.get('msg') if data else  '未知错误'}")

 return  None

# 使用示例

api = HKStockAPI()

quote = api.get_stock_quote("700")  # 腾讯控股

if quote:

 print(f"股票代码: {quote['s']}")

 print(f"最新价: {quote['ld']}")

 print(f"开盘价: {quote['o']}")

 print(f"最高价: {quote['h']}")

 print(f"最低价: {quote['l']}")

 print(f"成交量: {quote['v']}")

 print(f成交额: {quote['tu']}")

 # 时间戳转换

 import datetime

 trade_time = datetime.datetime.fromtimestamp(quote['t']/1000)

 print(f"交易时间: {trade_time}")

运行以上代码，将返回当前股票的行情信息，包括股票代码、最新价、开盘价、最高价、最低价、成交量、成交额和交易时间。

2. 股票实时盘口接口

盘口数据（深度数据）展示买卖双方的挂单情况，对于短线交易至关重要。

def  get_stock_depth(self, symbol, region="HK"):

 """

    获取港股实时盘口数据

    :param symbol: 股票代码

    :param region: 地区代码

    :return: 盘口数据字典

    """

endpoint = "/stock/depth"

    params = {

 "region": region,

 "code": symbol

    }

data = self._make_request(endpoint, params)

 if data and data.get("code") == 0:

 return data.get("data")

 else:

 print(f"获取盘口数据失败: {data.get('msg') if data else  '未知错误'}")

 return  None

# 使用示例

depth = api.get_stock_depth("700")  # 腾讯控股盘口

if depth:

 print(f"股票代码: {depth['s']}")

 print("=== 卖盘 ===")

 for ask in depth['a']:

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

 print("=== 买盘 ===")

 for bid in depth['b']:

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

 # 计算买卖盘压力

total_ask_volume = sum(ask['v'] for ask in depth['a'])

total_bid_volume = sum(bid['v'] for bid in depth['b'])

 print(f"买卖盘比例: {total_bid_volume/total_ask_volume:.2%}")

运行以上代码，将返回当前股票的买卖盘信息。

3. 获取股票历史数据

历史 K 线数据用于技术分析和策略回测，是量化交易的基石。

def  get_historical_kline(self, symbol, k_type=2, limit=100, region="HK",et):

 """

    获取港股历史K线数据

    :param symbol: 股票代码

    :param k_type: K线类型 (1: 1分钟, 2: 5分钟, 3: 15分钟, 4: 30分钟, 5: 60分钟, 6：2小时，7:4小时 8: 日K, 9: 周K, 10: 月K)

    :param limit: 数据条数

    :param region: 地区代码

    :param et: 查询截止时间戳 (为空时默认为当前时间戳)

    :return: K线数据列表

    """

endpoint = "/stock/kline"

    params = {

 "region": region,

 "code": symbol,

 "kType": k_type,

 "limit": limit,

 "et": et

    }

data = self._make_request(endpoint, params)

 if data and data.get("code") == 0:

 return data.get("data")

 else:

 print(f"获取K线数据失败: {data.get('msg') if data else  '未知错误'}")

 return  None

# 使用示例

kline_data = api.get_historical_kline("700", k_type=2, limit=50)

if kline_data:

 # 转换为Pandas DataFrame便于分析

 import pandas as pd

    df = pd.DataFrame(kline_data)

    df['t'] = pd.to_datetime(df['t'], unit='ms')

    df.set_index('t', inplace=True)

 print(df.head())

 # 计算简单移动平均

    df['sma_5'] = df['c'].rolling(window=5).mean()

    df['sma_20'] = df['c'].rolling(window=20).mean()

 # 绘制K线图

 import matplotlib.pyplot as plt

    plt.figure(figsize=(12, 6))

    plt.plot(df.index, df['c'], label='Close Price')

    plt.plot(df.index, df['sma_5'], label='5-day SMA')

    plt.plot(df.index, df['sma_20'], label='20-day SMA')

    plt.title('Tencent Holdings (0700.HK) Stock Price')

    plt.legend()

    plt.show()

4. WebSocket 实时数据推送

对于需要实时监控行情场景，WebSocket 比轮询方式更高效。

import websocket

import json

import threading

import time

class  HKStockWebSocket:

 def  __init__(self, token):

 self.ws_url = "wss://api.itick.org/stock"

 self.token = token

 self.ws = None

 self.is_connected = False

 def  on_message(self, ws, message):

 """处理WebSocket推送的消息"""

        data = json.loads(message)

 print(f"收到实时数据: {data}")

 # 根据数据类型进行不同处理

 if  'quote'  in data:

 self.handle_quote_data(data['quote'])

 elif  'depth'  in data:

 self.handle_depth_data(data['depth'])

 def  on_error(self, ws, error):

 print(f"WebSocket错误: {error}")

 def  on_close(self, ws, close_status_code, close_msg):

 self.is_connected = False

 print("WebSocket连接关闭")

 def  on_open(self, ws):

 """连接建立后的回调函数"""

 self.is_connected = True

 print("WebSocket连接已建立")

 # 订阅股票数据

        subscribe_msg = {

 "ac": "subscribe",

 "params": "700$HK,9988$HK",  # 腾讯控股、阿里巴巴

 "types": "depth,quote"

        }

        ws.send(json.dumps(subscribe_msg))

 def  handle_quote_data(self, quote_data):

 """处理实时报价数据"""

 print(f"股票 {quote_data['s']} 最新价: {quote_data['ld']}")

 def  handle_depth_data(self, depth_data):

 """处理盘口数据"""

 print(f"股票 {depth_data['s']} 盘口已更新")

 def  start(self):

 """启动WebSocket连接"""

 self.ws = websocket.WebSocketApp(

 self.ws_url,

 on_message=self.on_message,

 on_error=self.on_error,

 on_close=self.on_close,

 on_open=self.on_open,

 header=headers  # 使用统一的请求头

        )

 # 在单独线程中运行WebSocket

 def  run_ws():

 self.ws.run_forever()

 self.ws_thread = threading.Thread(target=run_ws)

 self.ws_thread.daemon = True

 self.ws_thread.start()

 def  stop(self):

 """停止WebSocket连接"""

 if  self.ws:

 self.ws.close()

# 使用示例

ws_client = HKStockWebSocket("yourtoken")

ws_client.start()

# 保持主线程运行

try:

 while  True:

        time.sleep(1)

except  KeyboardInterrupt:

    ws_client.stop()

 print("程序退出")

四、常见问题

1. 认证失败问题

问题现象：API 返回 401 或 403 状态码

解决方案：

• 检查 token 是否过期，及时刷新

• 验证请求头格式是否正确

• 确认 API 访问权限

2. 频率限制问题

问题现象：API 返回 429 状态码

解决方案：

• 升级到高级版

• 实现请求频率控制

• 合理安排数据更新间隔

• 使用批量接口减少请求次数

五、总结

港股实时行情 API 的接入是一个系统性工程，涉及接口选型、技术实现、性能优化等多个方面。通过本文的指南，您可以：

1. 根据需求选择合适的 API 服务商，平衡数据质量、实时性和成本

2. 快速实现核心行情接口的调用，包括实时报价、盘口数据、历史 K 线

3. 构建稳定的生产环境应用，通过缓存、重试、监控等机制提升系统可靠性

4. 有效处理常见问题，确保业务连续性和数据准确性

随着技术的不断发展，港股行情 API 也在持续进化。建议开发者持续关注各 API 提供商的更新动态，及时优化自己的实现方案，为量化交易和投资决策提供更加可靠的数据支持。

温馨提示：本文提供的代码示例仅供参考，请根据官方文档中提供的具体实现方式进行修改和测试。

github.com/itick-org

参考文档：https://docs.itick.org/rest-api/stocks/stock-kline

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