Python如何调用港股行情接口

伯绮梦 · 7 天前

1. 接口信息

接口类型：实时综合行情接口
支持品种：贵金属，商品期货，外汇，A股，港股，美股
查询方式：HTTP, WebSocket
申请密钥：https://infoway.io
官方对接文档：https://infoway.readme.io/reference/ws-subscription

复制代码

2. 获取股票清单

这个接口用来查询股票的名单，比如我可以获取美股清单：

import requests
url = "https://data.infoway.io/common/basic/symbols?type=STOCK_HK"
headers = {"accept": "application/json"}
response = requests.get(url, headers=headers)
print(response.text)

复制代码

只需要在type中传入STOCK_US，或者STOCK_CN即可查询所有美股或A股的清单。
3. 批量获取股票K线

这个接口用来获取多只股票的K线，请求地址：

https://data.infoway.io/stock/batch_kline/{klineType}/{klineNum}/{codes}
# {klineType} 是K线的时间
# 1 = 1分钟k线
# 2 = 5分钟k线
# 3 = 15分钟k线
# 4 = 30分钟k线
# 5 = 1小时k线
# 6 = 2小时k线
# 7 = 4小时k线
# 8 = 1日k线
# 9 = 1周k线
# 10 = 1月k线
# 11 = 1季k线
# 12 = 1年k线
# {klineNum}是需要返回的K线数量，最大支持一次性返回500根K线

复制代码

可以跨市场查询，比如同时获取A股，港股和美股的个股K线：

import requests
api_url = 'https://data.infoway.io/stock/batch_kline/1/10/002594.SZ%2C00285.HK%2CTSLA.US'
# 申请免费密钥: https://infoway.io
# 设置请求头
headers = {
'User-Agent': 'Mozilla/5.0',
'Accept': 'application/json',
'apiKey': 'yourApikey'
}
# 发送GET请求
response = requests.get(api_url, headers=headers)
# 输出结果
print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")

复制代码

4. 查询股票盘口

这个接口用于查询股票的交易盘口，最大支持5档盘口查询：

import requests
url = "https://data.infoway.io/stock/batch_depth/002594.SZ%2C00285.HK%2CTSLA.US%2CAUDCAD%2CHK50%2CXNIUSD"
# 申请免费密钥: https://infoway.io
# 设置请求头
headers = {
'User-Agent': 'Mozilla/5.0',
'Accept': 'application/json',
'apiKey': 'yourApikey'
}
response = requests.get(url, headers=headers)
print(response.text)

复制代码

5. WebSocket订阅实时股票行情

以上介绍的是HTTP请求，这种方式存在一定的延时，在实盘交易中推荐使用WebSocket来获取实时的行情推送。WebSocket的优势是和服务器建立一个长连接，只要连接不断开都会收到来自服务器的数据推送，延时一般在120ms以内。而你只需要每隔一分钟向服务器发送一个心跳来保活即可。下面是WebSocket的代码示例，虽然是查询的贵金属，但原理是一样的：

import json
import time
import schedule
import threading
import websocket
from loguru import logger
class WebsocketExample:
def __init__(self):
self.session = None
# 申请免费API Key: https://infoway.io
self.ws_url = "wss://data.infoway.io/ws?business=common&apikey=YourAPIKey"
self.reconnecting = False
self.last_ping_time = 0
self.max_ping_interval = 30 # 最大心跳包间隔时间
self.retry_attempts = 0 # 重试次数
self.max_retry_attempts = 5 # 最大重试次数
def connect_all(self):
"""建立WebSocket连接并启动自动重连机制"""
try:
self.connect(self.ws_url)
self.start_reconnection(self.ws_url)
except Exception as e:
logger.error(f"Failed to connect to {self.ws_url}: {str(e)}")
def start_reconnection(self, url):
"""启动定时重连检查"""
def check_connection():
if not self.is_connected():
logger.debug("Reconnection attempt...")
self.retry_attempts += 1
if self.retry_attempts <= self.max_retry_attempts:
self.connect(url)
else:
logger.error("Exceeded max retry attempts.")
# 使用线程定期检查连接状态
threading.Thread(target=lambda: schedule.every(10).seconds.do(check_connection), daemon=True).start()
def is_connected(self):
"""检查WebSocket连接状态"""
return self.session and self.session.connected
def connect(self, url):
"""建立WebSocket连接"""
try:
if self.is_connected():
self.session.close()
self.session = websocket.WebSocketApp(
url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
# 启动WebSocket连接（非阻塞模式）
threading.Thread(target=self.session.run_forever, daemon=True).start()
except Exception as e:
logger.error(f"Failed to connect to the server: {str(e)}")
def on_open(self, ws):
"""WebSocket连接建立成功后的回调"""
logger.info(f"Connection opened")
try:
# 发送贵金属实时成交明细订阅请求
trade_send_obj = {
"code": 10000,
"trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
"data": {"codes": "XAUUSD"} # XAUUSD 为贵金属黄金（黄金/美元）
}
self.send_message(trade_send_obj)
# 不同请求之间间隔一段时间
time.sleep(5)
# 发送贵金属实时盘口数据订阅请求
depth_send_obj = {
"code": 10003,
"trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
"data": {"codes": "XAUUSD"} # XAUUSD为黄金的实时盘口数据
}
self.send_message(depth_send_obj)
# 不同请求之间间隔一段时间
time.sleep(5)
# 发送贵金属实时K线数据订阅请求
kline_data = {
"arr": [
{
"type": 1,
"codes": "XAUUSD" # XAUUSD 为黄金K线数据
}
]
}
kline_send_obj = {
"code": 10006,
"trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
"data": kline_data
}
self.send_message(kline_send_obj)
# 启动定时心跳任务
threading.Thread(target=lambda: schedule.every(30).seconds.do(self.ping), daemon=True).start()
except Exception as e:
logger.error(f"Error sending initial messages: {str(e)}")
def on_message(self, ws, message):
"""接收消息的回调"""
try:
logger.info(f"Message received: {message}")
except Exception as e:
logger.error(f"Error processing message: {str(e)}")
def on_close(self, ws, close_status_code, close_msg):
"""连接关闭的回调"""
logger.info(f"Connection closed: {close_status_code} - {close_msg}")
def on_error(self, ws, error):
"""错误处理的回调"""
logger.error(f"WebSocket error: {str(error)}")
def send_message(self, message_obj):
"""发送消息到WebSocket服务器"""
if self.is_connected():
try:
self.session.send(json.dumps(message_obj))
except Exception as e:
logger.error(f"Error sending message: {str(e)}")
else:
logger.warning("Cannot send message: Not connected")
def ping(self):
"""发送心跳包"""
current_time = time.time()
if current_time - self.last_ping_time >= self.max_ping_interval:
ping_obj = {
"code": 10010,
"trace": "01213e9d-90a0-426e-a380-ebed633cba7a"
}
self.send_message(ping_obj)
self.last_ping_time = current_time
else:
logger.debug(f"Ping skipped: Time interval between pings is less than {self.max_ping_interval} seconds.")
# 使用示例
if __name__ == "__main__":
ws_client = WebsocketExample()
ws_client.connect_all()
# 保持主线程运行
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
logger.info("Exiting...")
if ws_client.is_connected():
ws_client.session.close()

复制代码

6. 常见问题

HTTP和WebSocket有什么区别？

通过HTTP请求，返回的数据是基于请求和响应传输的。客户端向服务器发送请求，服务器返回响应。每次请求/响应都必须是完整的。而WebSocket的数据是全双工的，客户端和服务器可以随时向对方发送数据，也就是所谓的长连接。它更适用于实时通信，因为一旦连接建立，可以保持双向流量。
港股的延迟是多少？

这个和接口的设计有关，一般通过WebSocket获取的数据，延迟在100ms左右。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

Python如何调用港股行情接口

浏览过的版块

签约作者