欧洲股票实时行情与历史数据:python接入指南

原创

FxStock Lab

修改于 2025-11-29 18:02:27

如何接入欧洲股票实时行情与历史数据：德国、荷兰、英国、意大利

在全球金融市场日益互联的今天，欧洲股票市场作为经济引擎之一，吸引了无数投资者和开发者。德国 DAX 指数成分股、荷兰 AEX 指数股票、英国 FTSE 100 上市公司，还是意大利 MIB 指数等市场以其成熟的交易体系和活跃的股票品种闻名。但如何高效接入这些市场的实时行情和历史数据？本文以技术实现为视角，介绍通过 WebSocket 和 REST API 接入相关市场数据的通用思路及示例代码，通过低延时数据 API、股票接口和实时行情接口，帮助您轻松获取金融行情，包括股票 tick 数据和全面的股票 API 接口，供技术开发者参考。

实时行情接入：WebSocket API 技术实现

实时行情通常通过 WebSocket 协议实现低延时数据推送，支持成交（tick）、报价（quote）和盘口（depth）等 Level 1/Level 2 数据。以欧洲主要市场（德国 DE、荷兰 NL、英国 GB、意大利 IT）为例，接入流程一般包括连接建立、认证、标的订阅及心跳维护。

WebSocket API 参数说明

参数名称	描述	类型	必填
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"
  }
}

历史数据接入：使用 REST API 查询 K 线

历史数据（如 K 线）通常通过 REST API 查询，支持分钟线、日线等多周期，包含开盘价（o）、最高价（h）、最低价（l）、收盘价（c）、成交量（v）等字段。欧洲市场历史数据的完整性及复权准确性需以接口服务商的说明为准。

REST API 参数说明

REST API 的 K 线查询接口通常包含以下参数（需参考具体官网文档）：

入参示例：

参数名称	描述	类型	必填	示例
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

出参示例（响应数据结构）：

参数名称	描述	类型	示例
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 线数据
  ]
}