如何实现实现量化分析,首先获取股票实时行情、股票历史数据和股票行情数据是进行量化交易和分析的关键。通过可靠的股票实时行情接口,如股票 API ,股票实时报价 API 和股票行情 api ,开发者可以轻松接入全球市场数据。本文将介绍如何使用专业的股票实时报价 API 、金融 api 和金融行情数据 API 来对接德国股票行情,特别是法兰克福交易所( FWB/Xetra ),从而实现高效的量化分析。这些工具不仅提供毫秒级延迟的实时数据,还支持历史回测,帮助投资者做出数据驱动的决策。
法兰克福交易所( FWB/Xetra )是欧洲最大的股票交易所之一,涵盖了众多德国蓝筹股,如阿迪达斯( ADS )、德意志银行( DBK )等。它以高效的电子交易系统闻名,交易量巨大,适合量化策略的开发。通过 API 接入,我们可以获取实时报价、历史 K 线和盘口深度数据,这些数据是构建均线策略、波动率分析等量化模型的基础。
在量化交易领域,选择一个合适的股票数据 API 对策略的成败至关重要。对于德国股票市场,尤其是法兰克福交易所,开发者通常面临三个核心挑战:数据时效性、完整性和合规性要求
市场上主要有几种 API 解决方案:
iTick 作为聚焦欧洲市场的金融数据服务商,其 API 实现了法兰克福交易所全品种覆盖(含 XETRA 交易品种),支持毫秒级股票实时行情推送与 20 年历史分笔数据获取,完全适配 MiFID II 监管要求,还提供 Python SDK 与完整的量化工具集成方案,注册既可享受免费开发套餐,适合中高频策略与深度量化分析
Alpha Vantage 支持包括德国 DAX 指数成分股在内的全球 30 多个国家股票数据,免费版每日支持 500 次调用。但其主要限制在于德国股票实时 API 延迟长达 15 分钟(非付费用户),且历史数据仅提供 10 年日线级别,无 Level 2 深度行情。
IEX Cloud 提供法兰克福交易所实时股票报价 API ,延迟约为 1 秒,并整合了财务报表与 ESG 数据。但它对德国股票的覆盖仅限于 DAX30 成分股,历史数据最长只有 5 年
提示:无论选择哪种 API ,都需先完成平台注册与认证,获取专属 API 密钥( Key ),这是接口调用的身份凭证,需妥善保管避免泄露。
本文参考 iTick API ,这是一个支持全球多个市场的金融数据接口,包括德国( region=DE )。它提供 RESTful API 和 WebSocket 两种方式,数据覆盖实时报价、历史 K 线和盘口深度。注意:使用前需注册账号并获取 token ,本文代码中的"your_token"需替换为实际值。
首先,访问 iTick 官网注册账号,获取 API Token 。该 API 支持的 region 包括 DE (德国),code 为股票符号(如 ADS 为阿迪达斯)。测试时,确保你的订阅计划支持德国市场数据。
实时报价 API 提供最新价、开盘价、最高价、最低价等核心指标。接口路径:GET /stock/quote?region={region}&code={code}
Python 代码示例:
import requests
url = "https://api.itick.org/stock/quote?region=DE&code=ADS"
headers = {
"accept": "application/json",
"token": "your_token"
}
response = requests.get(url, headers=headers)
data = response.json()
if data["code"] == 0:
quote = data["data"]
print(f"股票代码: {quote['s']}")
print(f"最新价: {quote['ld']}")
print(f"开盘价: {quote['o']}")
print(f"最高价: {quote['h']}")
print(f"最低价: {quote['l']}")
print(f"涨跌幅: {quote['chp']}%")
else:
print("请求失败:", data["msg"])
这个接口返回的 JSON 数据结构清晰,便于解析。在量化分析中,你可以用最新价计算实时收益率。
历史 K 线是量化回测的核心,支持分钟级到月级周期。接口路径:GET /stock/kline?region={region}&code={code}&kType={kType}&limit={limit}
例如,获取阿迪达斯最近 100 根 日 K 线:
import requests
import pandas as pd
from datetime import datetime
def fetch_historical_data(symbol, region="DE", kType=8, limit=100):
"""
获取历史 K 线数据
参数:
symbol: 股票代码,如"ADS"
region: 市场代码,德国为"DE"
kType: K 线类型,1-分钟线,2-5 分钟线,8-日线,9-周线,10-月线
limit: 获取的数据条数
"""
url = f"https://api.itick.org/stock/kline?region={region}&code={symbol}&kType={kType}&limit={limit}"
headers = {
"accept": "application/json",
"token": "your_token" # 替换为实际 Token
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status() # 检查请求是否成功
data = response.json()
if data.get("code") == 0 and "data" in data:
# 将数据转换为 Pandas DataFrame
df = pd.DataFrame(data["data"])
# 转换时间戳为可读格式
df['datetime'] = pd.to_datetime(df['t'], unit='ms')
# 设置列为标准金融数据格式
df.rename(columns={
'o': 'Open',
'h': 'High',
'l': 'Low',
'c': 'Close',
'v': 'Volume',
'tu': 'Turnover'
}, inplace=True)
# 选择并排序列
df = df[['datetime', 'Open', 'High', 'Low', 'Close', 'Volume', 'Turnover']]
df.set_index('datetime', inplace=True)
return df
else:
print(f"获取数据失败: {data.get('msg')}")
return None
except requests.exceptions.RequestException as e:
print(f"请求错误: {e}")
return None
def analyze_german_stocks():
"""分析多只德国股票的历史表现"""
symbols = ["ADS", "SAP", "VOW3", "ALV", "MRK"]
all_data = {}
for symbol in symbols:
print(f"正在获取{symbol}的历史数据...")
df = fetch_historical_data(symbol, kType=8, limit=200) # 获取 200 条日线数据
if df is not None and len(df) > 0:
all_data[symbol] = df
# 计算基本统计指标
latest_close = df['Close'].iloc[-1]
previous_close = df['Close'].iloc[-2] if len(df) > 1 else latest_close
daily_change = ((latest_close - previous_close) / previous_close * 100) if len(df) > 1 else 0
# 计算 20 日移动平均
ma_20 = df['Close'].rolling(window=20).mean().iloc[-1]
print(f"{symbol}:")
print(f" 最新收盘价: {latest_close:.2f}欧元")
print(f" 日涨跌幅: {daily_change:+.2f}%")
print(f" 20 日移动平均: {ma_20:.2f}欧元")
print(f" 数据时间范围: {df.index[0].date()} 至 {df.index[-1].date()}")
print()
return all_data
if __name__ == "__main__":
# 获取并分析德国股票数据
stock_data = analyze_german_stocks()
# 如果获取到了数据,可以进行进一步分析
if stock_data:
print("数据获取完成,可以进行量化策略回测和分析了!")
这有助于识别趋势反转点。
盘口深度提供买卖五档或十档数据,反映市场挂单情况。接口路径:GET /stock/depth?region={region}&code={code}
import requests
url = "https://api.itick.org/stock/depth?region=DE&code=ADS"
headers = {
"accept": "application/json",
"token": "your_token"
}
response = requests.get(url, headers=headers)
data = response.json()
if data["code"] == 0:
depth = data["data"]
print(f"股票代码: {depth['s']}")
print("卖盘:")
for ask in depth['a'][:5]: # 显示前 5 档卖盘
print(f"档位{ask['po']}: 价格 {ask['p']}, 挂单量 {ask['v']}, 订单数 {ask['o']}")
print("买盘:")
for bid in depth['b'][:5]: # 显示前 5 档买盘
print(f"档位{bid['po']}: 价格 {bid['p']}, 挂单量 {bid['v']}, 订单数 {bid['o']}")
else:
print("请求失败:", data["msg"])
在量化中,盘口数据可用于计算买卖压力比,帮助判断市场情绪。
对于高频量化,RESTful API 可能有延迟,推荐 WebSocket 。连接后订阅数据,支持 tick 、quote 和 depth 类型。
Python 示例(使用 websocket 库):
import websocket
import json
import threading
import time
# WebSocket 连接地址和 Token
WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_token" # 替换为实际 Token
def on_message(ws, message):
"""处理接收到的消息"""
data = json.loads(message)
# 处理连接成功的消息
if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
print("连接成功,等待认证...")
# 处理认证结果
elif data.get("resAc") == "auth":
if data.get("code") == 1:
print("认证成功")
subscribe(ws) # 认证成功后订阅数据
else:
print("认证失败")
ws.close()
# 处理订阅结果
elif data.get("resAc") == "subscribe":
if data.get("code") == 1:
print("订阅成功")
else:
print("订阅失败:", data.get("msg"))
# 处理市场数据
elif data.get("data"):
market_data = data["data"]
data_type = market_data.get("type")
symbol = market_data.get("s")
if data_type == "tick":
print(f"成交数据 {symbol}: 最新价={market_data['ld']}, 成交量={market_data['v']}, 时间={market_data['t']}")
elif data_type == "quote":
print(f"报价数据 {symbol}: 开={market_data['o']}, 高={market_data['h']}, 低={market_data['l']}, 收={market_data['ld']}")
elif data_type == "depth":
print(f"盘口数据 {symbol}: 买一价={market_data['b'][0]['p'] if market_data['b'] else 'N/A'}, "
f"卖一价={market_data['a'][0]['p'] if market_data['a'] else 'N/A'}")
def on_error(ws, error):
"""处理错误"""
print("错误:", error)
def on_close(ws, close_status_code, close_msg):
"""连接关闭回调"""
print("连接关闭")
def on_open(ws):
"""连接建立后的回调"""
print("WebSocket 连接已打开")
def subscribe(ws):
"""订阅行情数据"""
subscribe_msg = {
"ac": "subscribe",
# 订阅德国 Adidas 、SAP 和大众汽车的实时数据
"params": "ADS$DE,SAP$DE,VOW3$DE",
"types": "tick,quote,depth" # 订阅成交、报价和盘口数据
}
ws.send(json.dumps(subscribe_msg))
print("订阅消息已发送")
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("心跳包已发送")
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()
这段代码建立了与 iTick WebSocket 服务器的连接,并订阅了德国三家知名公司( Adidas 、SAP 和大众汽车)的实时数据。连接建立后,服务器会持续推送三种类型的数据:
通过 WebSocket 获取实时数据的优势在于低延迟和高效的数据推送机制,特别适合需要实时监控市场并快速做出交易决策的量化策略
获取数据只是第一步,真正的价值在于如何利用这些数据进行量化分析。下面我们结合实时数据和历史数据,构建一个简单的量化分析示例。
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
class GermanStockAnalyzer:
"""德国股票分析器"""
def __init__(self, historical_data):
self.data = historical_data
def calculate_technical_indicators(self):
"""计算常见技术指标"""
df = self.data.copy()
# 计算移动平均线
df['MA_5'] = df['Close'].rolling(window=5).mean()
df['MA_20'] = df['Close'].rolling(window=20).mean()
df['MA_60'] = df['Close'].rolling(window=60).mean()
# 计算相对强弱指数(RSI)
delta = df['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# 计算布林带
df['BB_middle'] = df['Close'].rolling(window=20).mean()
bb_std = df['Close'].rolling(window=20).std()
df['BB_upper'] = df['BB_middle'] + 2 * bb_std
df['BB_lower'] = df['BB_middle'] - 2 * bb_std
# 计算成交量加权平均价格(VWAP) - 日内指标
df['VWAP'] = (df['Turnover'] / df['Volume']).rolling(window=20).mean()
return df
def generate_signals(self, df):
"""基于技术指标生成交易信号"""
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
signals['signal'] = 0
# 双移动平均线交叉策略
signals['ma_signal'] = 0
signals.loc[df['MA_5'] > df['MA_20'], 'ma_signal'] = 1 # 金叉
signals.loc[df['MA_5'] < df['MA_20'], 'ma_signal'] = -1 # 死叉
# RSI 超买超卖信号
signals['rsi_signal'] = 0
signals.loc[df['RSI'] < 30, 'rsi_signal'] = 1 # 超卖,买入信号
signals.loc[df['RSI'] > 70, 'rsi_signal'] = -1 # 超买,卖出信号
# 布林带突破信号
signals['bb_signal'] = 0
signals.loc[df['Close'] < df['BB_lower'], 'bb_signal'] = 1 # 突破下轨,买入信号
signals.loc[df['Close'] > df['BB_upper'], 'bb_signal'] = -1 # 突破上轨,卖出信号
# 综合信号
signals['combined_signal'] = signals[['ma_signal', 'rsi_signal', 'bb_signal']].mean(axis=1)
# 生成最终交易信号
signals.loc[signals['combined_signal'] > 0.3, 'signal'] = 1 # 强烈买入
signals.loc[signals['combined_signal'] < -0.3, 'signal'] = -1 # 强烈卖出
return signals
def plot_analysis(self, df, signals):
"""可视化分析结果"""
fig, axes = plt.subplots(3, 1, figsize=(15, 12))
# 价格与移动平均线
ax1 = axes[0]
ax1.plot(df.index, df['Close'], label='收盘价', linewidth=1)
ax1.plot(df.index, df['MA_5'], label='5 日 MA', linewidth=1, alpha=0.7)
ax1.plot(df.index, df['MA_20'], label='20 日 MA', linewidth=1, alpha=0.7)
ax1.plot(df.index, df['MA_60'], label='60 日 MA', linewidth=1, alpha=0.7)
# 标记交易信号
buy_signals = signals[signals['signal'] == 1]
sell_signals = signals[signals['signal'] == -1]
ax1.scatter(buy_signals.index, df.loc[buy_signals.index, 'Close'],
color='green', marker='^', s=100, label='买入信号')
ax1.scatter(sell_signals.index, df.loc[sell_signals.index, 'Close'],
color='red', marker='v', s=100, label='卖出信号')
ax1.set_title('德国股票价格与移动平均线')
ax1.set_ylabel('价格(欧元)')
ax1.legend()
ax1.grid(True, alpha=0.3)
# RSI 指标
ax2 = axes[1]
ax2.plot(df.index, df['RSI'], label='RSI', linewidth=1, color='purple')
ax2.axhline(y=70, color='red', linestyle='--', alpha=0.5, label='超买线')
ax2.axhline(y=30, color='green', linestyle='--', alpha=0.5, label='超卖线')
ax2.fill_between(df.index, 30, 70, alpha=0.1, color='gray')
ax2.set_title('相对强弱指数(RSI)')
ax2.set_ylabel('RSI 值')
ax2.legend()
ax2.grid(True, alpha=0.3)
# 布林带
ax3 = axes[2]
ax3.plot(df.index, df['Close'], label='收盘价', linewidth=1)
ax3.plot(df.index, df['BB_middle'], label='中轨', linewidth=1, alpha=0.7)
ax3.plot(df.index, df['BB_upper'], label='上轨', linewidth=1, alpha=0.7, linestyle='--')
ax3.plot(df.index, df['BB_lower'], label='下轨', linewidth=1, alpha=0.7, linestyle='--')
ax3.fill_between(df.index, df['BB_lower'], df['BB_upper'], alpha=0.1)
ax3.set_title('布林带')
ax3.set_ylabel('价格(欧元)')
ax3.set_xlabel('日期')
ax3.legend()
ax3.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
def backtest_strategy(self, signals, initial_capital=10000):
"""简单策略回测"""
capital = initial_capital
position = 0
trades = []
for i in range(1, len(signals)):
current_price = signals['price'].iloc[i]
signal = signals['signal'].iloc[i]
if signal == 1 and position == 0: # 买入信号,且当前无持仓
position = capital / current_price
capital = 0
trades.append({
'date': signals.index[i],
'action': 'BUY',
'price': current_price,
'position': position
})
elif signal == -1 and position > 0: # 卖出信号,且当前有持仓
capital = position * current_price
position = 0
trades.append({
'date': signals.index[i],
'action': 'SELL',
'price': current_price,
'capital': capital
})
# 计算最终收益
if position > 0:
final_capital = position * signals['price'].iloc[-1]
else:
final_capital = capital
total_return = (final_capital - initial_capital) / initial_capital * 100
return {
'initial_capital': initial_capital,
'final_capital': final_capital,
'total_return': total_return,
'trades': trades
}
# 使用示例
if __name__ == "__main__":
# 假设我们已经获取了历史数据
# 这里使用模拟数据演示
dates = pd.date_range(start='2024-01-01', end='2024-12-01', freq='D')
np.random.seed(42)
prices = 100 + np.cumsum(np.random.randn(len(dates)) * 0.5)
volumes = np.random.randint(100000, 1000000, len(dates))
historical_data = pd.DataFrame({
'Close': prices,
'Volume': volumes,
'Turnover': prices * volumes
}, index=dates)
# 创建分析器实例
analyzer = GermanStockAnalyzer(historical_data)
# 计算技术指标
df_with_indicators = analyzer.calculate_technical_indicators()
# 生成交易信号
signals = analyzer.generate_signals(df_with_indicators)
# 可视化分析
analyzer.plot_analysis(df_with_indicators, signals)
# 回测策略
backtest_result = analyzer.backtest_strategy(signals)
print("策略回测结果:")
print(f"初始资金: {backtest_result['initial_capital']:.2f}欧元")
print(f"最终资金: {backtest_result['final_capital']:.2f}欧元")
print(f"总收益率: {backtest_result['total_return']:.2f}%")
print(f"交易次数: {len(backtest_result['trades'])}")
这个量化分析示例展示了如何将从 iTick API 获取的数据应用于实际的量化策略中。通过计算技术指标、生成交易信号和进行策略回测,我们可以系统性地评估交易策略的有效性。
通过 Python 对接法兰克福交易所 API 股票实时行情与历史数据,我们搭建了量化分析的核心数据管道。这不仅是技术的实现,更是以数据驱动决策的开始——稳定可靠的数据流让策略回测更精准、信号生成更及时,为在严谨的欧洲市场中探索 alpha 机会奠定了坚实基础。现在,您已拥有连接全球重要金融市场的能力,是时候将这些数据转化为您的策略优势了。
原文地址: https://itick.org/blog/stock-api/free-german-stock-api-comparison