如何接入欧洲股票实时行情与历史数据:德国、荷兰、英国、意大利
如何接入欧洲股票实时行情与历史数据:德国、荷兰、英国、意大利
在全球金融市场日益互联的今天,欧洲股票市场作为经济引擎之一,吸引了无数投资者和开发者。德国(DE)、荷兰(NL)、英国(GB)和意大利(IT)等市场以其成熟的交易体系和活跃的股票品种闻名。但如何高效接入这些市场的实时行情和历史数据?本文介绍一款专业的股票行情接口提供商,通过低延时数据 API、股票接口和实时行情接口,帮助您轻松获取金融行情,包括股票 tick 数据和全面的股票 API 接口。今天,我们将详细指导您如何使用 WebSocket 和 REST API 接入欧洲股票的实时与历史数据,让您的量化策略、交易应用或数据分析更具竞争力。
多个市场支持,包括欧洲的核心交易所。无论是追踪德国 DAX 指数成分股、荷兰 AEX 指数股票、英国 FTSE 100 上市公司,还是意大利 MIB 指数下的蓝筹股,他的的股票 api 接口都能提供毫秒级推送和历史查询,确保数据准确、低延时。接下来,我们分步解析接入流程,并提供 Python 代码示例、参数表格、入参/出参示例以及响应示例。
实时行情接入:使用 WebSocket API 获取股票 tick 和报价
WebSocket API 是实时行情接口的核心,支持 Level 1 & Level 2 数据推送,包括成交(tick)、报价(quote)和盘口(depth)。对于欧洲市场,您可以订阅德国(region=DE)、荷兰(NL)、英国(GB)和意大利(IT)的股票代码,实现低延时数据 API 接入。连接过程包括连接、验证、订阅和心跳维护。
WebSocket API 参数说明
WebSocket 连接 URL:wss://api.itick.org/stock
| 参数名称 | 描述 | 类型 | 必填 |
|---|---|---|---|
| token | API 密钥,用于授权连接 | String | 是 |
订阅请求(JSON 格式):
| 参数名称 | 描述 | 类型 | 必填 | 示例 |
|---|---|---|---|---|
| ac | 操作类型,固定为 “subscribe” | String | 是 | “subscribe” |
| params | 标的代码,支持多个,用逗号分隔。格式:code$region(例如,德国股票:ADS$DE) | String | 是 | "ADS$DE,UNA$NL,VOD$GB,ENI$IT" |
| types | 订阅类型,支持 “tick”(成交)、”quote”(报价)、”depth”(盘口),用逗号分隔 | String | 是 | “tick,quote,depth” |
入参示例(订阅德国 Adidas、荷兰 Unilever、英国 Vodafone 和意大利 Eni 的实时数据):
{
"ac": "subscribe",
"params": "ADS$DE,UNA$NL,VOD$GB,ENI$IT",
"types": "tick,quote,depth"
}
出参示例(响应数据结构):
| 参数名称 | 描述 | 类型 | 示例 |
|---|---|---|---|
| code | 响应代码,1 表示成功 | Integer | 1 |
| msg | 消息描述 | String | “subscribe Successfully” |
| data | 数据内容,包含 s(符号)、ld(最新价)、v(成交量)等,根据 types 不同而变 | Object | 详见响应示例 |
Python 代码示例:接入实时行情
以下是使用 Python 的 WebSocket 客户端示例,订阅欧洲股票的股票 tick、报价和盘口数据。确保替换 your_token 为您的 API 密钥。
import websocket
import json
import threading
import time
# WebSocket 连接地址和 token
WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_token"
def on_message(ws, message):
"""处理接收到的消息"""
print("Received message:", message)
data = json.loads(message)
# 处理连接成功的消息
if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
print("Connected successfully, waiting for authentication...")
# 处理认证结果
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")
print(f"{data_type.upper()} data for {symbol}:", market_data)
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):
"""订阅行情数据"""
subscribe_msg = {
"ac": "subscribe",
"params": "ADS$DE,UNA$NL,VOD$GB,ENI$IT",
"types": "tick,quote,depth"
}
ws.send(json.dumps(subscribe_msg))
print("Subscribe message sent")
def send_ping(ws):
"""定期发送心跳包"""
while True:
time.sleep(30) # 每30秒发送一次心跳
ping_msg = {
"ac": "ping",
"params": str(int(time.time() * 1000))
}
ws.send(json.dumps(ping_msg))
print("Ping sent")
if __name__ == "__main__":
# 创建 WebSocket 连接,通过header传递token
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()
响应示例
成交(tick)响应:
{
"code": 1,
"data": {
"s": "ADS.DE",
"ld": 225.215,
"v": 16742235,
"t": 1731689407000,
"type": "tick"
}
}
报价(quote)响应:
{
"code": 1,
"data": {
"s": "ADS.DE",
"ld": 225.215,
"o": 226.27,
"h": 226.92,
"l": 224.44,
"t": 1731689407000,
"v": 16742235,
"tu": 3774688301.452,
"ts": 0,
"type": "quote"
}
}
盘口(depth)响应:
{
"code": 1,
"data": {
"s": "ADS.DE",
"a": [
{ "po": 1, "p": 3034.01, "v": 10.6023, "o": 10.6023 }
// ... 更多卖盘
],
"b": [
{ "po": 1, "p": 3034, "v": 20.9758, "o": 20.9758 }
// ... 更多买盘
],
"type": "depth"
}
}
通过这个实时行情接口,您可以构建低延时交易系统,实时监控欧洲股票的金融行情和股票 tick 变化。
历史数据接入:使用 REST API 查询 K 线
对于历史分析股票接口的 K 线查询 API,支持从分钟线到月线的多周期数据。欧洲市场如德国、荷兰、英国和意大利的股票历史数据完整,包括 OHLCV(开盘、最高、最低、收盘、成交量)和成交额,确保复权准确。
REST API 参数说明
请求 URL:GET https://api.itick.org/stock/kline?region={region}&code={code}&kType={kType}&limit={limit}&et={et}
| 参数名称 | 描述 | 类型 | 必填 | 示例 |
|---|---|---|---|---|
| region | 市场代码,支持 DE(德国)、NL(荷兰)、GB(英国)、IT(意大利)等 | String | 是 | “DE” |
| code | 产品代码 | String | 是 | “ADS” |
| kType | K 线类型(1:1 分钟, 2:5 分钟, 3:15 分钟, 4:30 分钟, 5:1 小时, 6:2 小时, 7:4 小时, 8:日, 9:周, 10:月) | Integer | 是 | 2 |
| limit | K 线数量 | Integer | 是 | 10 |
| et | 截止时间戳(为空默认为当前) | Integer | 否 | 1741239000000 |
入参示例(查询德国 Adidas 的最近 10 条 5 分钟 K 线):
https://api.itick.org/stock/kline?region=DE&code=ADS&kType=2&limit=10
出参示例(响应数据结构):
| 参数名称 | 描述 | 类型 | 示例 |
|---|---|---|---|
| t | 时间戳 | Number | 1741239000000 |
| o | 开盘价 | Number | 535 |
| h | 最高价 | Number | 536 |
| l | 最低价 | Number | 534.5 |
| c | 收盘价 | Number | 534.5 |
| v | 成交数量 | Number | 104799385 |
| tu | 成交额 | Number | 56119888070.5 |
Python 代码示例:查询历史 K 线
以下是使用 Python 的 REST 请求示例,查询欧洲股票的历史数据。确保替换 your_token 为您的 API 密钥。
import requests
url = "https://api.itick.org/stock/kline?region=DE&code=ADS&kType=2&limit=10"
headers = {
"accept": "application/json",
"token": "your_token"
}
response = requests.get(url, headers=headers)
print(response.text)
响应示例
{
"code": 0,
"msg": null,
"data": [
{
"tu": 56119888070.5,
"c": 534.5,
"t": 1741239000000,
"v": 104799385,
"h": 536,
"l": 534.5,
"o": 535
}
// ... 更多 K 线数据
]
}
这个股票 api 接口适合回测策略或构建图表,提供全面的金融行情历史洞察。
结语
欧洲股市作为全球金融市场的重要组成部分,掌握实时行情数据的获取方式,是你踏入这片投资天地的第一步,也是构建成功投资策略的基石。无论是实时行情监控、高频 Tick 分析还是历史策略回测,你都可以在这里找到适合的解决方案轻松接入德国、荷兰、英国和意大利等欧洲市场的低延时数据 API、股票 tick 和历史 K 线。
注:本文提供的代码示例仅供参考,实际使用请根据官方最新文档修改
GitHub:https://github.com/itick-org/
本作品采用《CC 协议》,转载必须注明作者和本文链接
关于 LearnKu