港股實時行情 API 接口接入實戰指南:從選型到落地全解析
在量化交易和金融科技應用開發中,實時行情數據是決策的核心。對於港股市場而言,選擇合適的數據接口並正確接入,是構建穩定交易系統的第一步。
隨着港股市場與 A 股市場的互聯互通機制不斷深化,港股實時行情數據的需求日益增長。無論是構建量化交易系統、開發投資分析工具,還是打造金融資訊平台,都離不開穩定可靠的行情數據接口。
本文將全面解析港股實時行情 API 接口的選型與接入過程,提供從主流接口對比、Python 實戰代碼到生產環境部署的完整指南。
一、港股行情 API 接口選型指南
目前市場上港股行情 API 魚龍混雜,從傳統金融數據服務商到新興技術平台,各有優劣。我們從開發者最關注的實時性、成本、易用性、數據完整性四個維度,對比 itick 與兩款主流接口的核心差異:
| API 服務商 | 實時性 | 接入成本 | 易用性 | 核心優勢 | 適用場景 |
|---|---|---|---|---|---|
| iTick | 毫秒級推送,延遲 ≤50ms | 免費版支持基礎行情,付費版性價比高 | RESTful 規範,請求頭統一,返回格式簡潔 | 港股全量覆蓋,支持 WebSocket 實時訂閲 | 個人開發者、量化交易、中小型行情繫統 |
| Wind 金融終端 API | 延遲 100-300ms | 年費高昂(數萬元起),無免費版 | 需安裝客户端,接口參數複雜 | 數據維度極全,含深度研報 | 大型金融機構、專業投研團隊 |
| Tushare 港股接口 | 延遲 1-3 秒,非實時 | 積分制,高頻調用需付費兑換積分 | Python SDK 友好,但港股數據覆蓋不全 | A 股數據優質,文檔完善 | A 股為主、港股為輔的低頻分析場景 |
二、接入前置準備:統一 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 的接入是一個系統性工程,涉及接口選型、技術實現、性能優化等多個方面。通過本文的指南,您可以:
- 根據需求選擇合適的 API 服務商,平衡數據質量、實時性和成本
- 快速實現核心行情接口的調用,包括實時報價、盤口數據、歷史 K 線
- 構建穩定的生產環境應用,通過緩存、重試、監控等機制提升系統可靠性
- 有效處理常見問題,確保業務連續性和數據準確性
隨着技術的不斷髮展,港股行情 API 也在持續進化。建議開發者持續關注各 API 提供商的更新動態,及時優化自己的實現方案,為量化交易和投資決策提供更加可靠的數據支持。
温馨提示:本文提供的代碼示例僅供參考,請根據官方文檔中提供的具體實現方式進行修改和測試。
https://github.com/itick-org
參考文檔:https://docs.itick.org/rest-api/stocks/stock-kline