如何接入欧洲股票实时行情与历史数据：德国、荷兰、英国、意大利
在全球金融市场日益互联的今天，欧洲股票市场作为经济引擎之一，吸引了无数投资者和开发者。德国（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 | 标的代码，支持多个，用逗号分隔。格式：coderegion（例如，德国股票：ADSDE） | 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 密钥。
```python
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 = "//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 线。
注：本文提供的代码示例仅供参考，实际使用请根据官方最新文档修改
参考文档：https://docs.itick.org/
GitHub：https://github.com/itick-org/