To achieve quantitative analysis, first obtaining real-time stock quotes, historical stock data, and market data is key to quantitative trading and analysis. Through reliable stock real-time quote interfaces, such as stock APIs, stock real-time quote APIs, and stock market APIs, developers can easily access global market data. This article will introduce how to use professional stock real-time quote APIs, financial APIs, and financial market data APIs to connect to German stock market data, particularly the Frankfurt Stock Exchange (FWB/Xetra), thereby enabling efficient quantitative analysis. These tools not only provide millisecond-level low-latency real-time data but also support historical backtesting, helping investors make data-driven decisions.
API Integration Scheme Comparison
The Frankfurt Stock Exchange (FWB/Xetra) is one of Europe's largest stock exchanges, covering numerous German blue-chip stocks such as Adidas (ADS) and Deutsche Bank (DBK). It is renowned for its efficient electronic trading system and massive trading volume, making it suitable for the development of quantitative strategies. Through API integration, we can obtain real-time quotes, historical candlestick charts, and order book depth data, which form the foundation for building quantitative models such as moving average strategies and volatility analysis.
In the field of quantitative trading, selecting an appropriate stock data API is crucial to the success of strategies. For the German stock market, especially the Frankfurt Stock Exchange, developers typically face three core challenges: data timeliness, completeness, and compliance requirements.
There are several main API solutions available in the market:
iTick, as a financial data provider focused on the European market, has its API achieving full coverage of Frankfurt Stock Exchange instruments (including XETRA-traded varieties), supporting millisecond-level real-time stock quote pushes and 20 years of historical tick data retrieval, fully compliant with MiFID II regulatory requirements. It also provides Python SDK and complete quantitative tool integration solutions. Registration allows access to a free development package, suitable for medium-to-high-frequency strategies and in-depth quantitative analysis.
Alpha Vantage supports stock data from over 30 countries globally, including components of the German DAX index, with the free version allowing 500 calls per day. However, its main limitations for German stock real-time APIs include delays of up to 15 minutes (for non-paid users), and historical data only provides 10 years of daily-level data without Level 2 depth quotes.
IEX Cloud provides real-time stock quote APIs for the Frankfurt Stock Exchange with approximately 1-second delays and integrates financial statements and ESG data. However, its coverage of German stocks is limited to DAX30 components, and historical data extends only up to 5 years.
Tip: Regardless of the API chosen, you must first complete platform registration and authentication to obtain a dedicated API key (Key). This is the credential for interface calls and should be securely stored to prevent leakage.
Preparation: Obtaining API Token
This article references the iTick API, which is a financial data interface supporting multiple global markets, including Germany (region=DE). It provides both RESTful API and WebSocket methods, with data covering real-time quotes, historical candlesticks, and order book depth. Note: Before use, register an account and obtain a token. In the code examples in this article, replace "your_token" with the actual value.
First, visit the iTick official website to register an account and obtain the API Token. The API supports regions including DE (Germany), with codes as stock symbols (e.g., ADS for Adidas). During testing, ensure your subscription plan supports German market data.
Step 1: Obtaining Real-Time Quotes (Quote)
The real-time quote API provides core indicators such as the latest price, opening price, highest price, and lowest price. Interface path: GET /stock/quote?region={region}&code={code}
Python code example:
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"Stock Code: {quote['s']}")
print(f"Latest Price: {quote['ld']}")
print(f"Opening Price: {quote['o']}")
print(f"Highest Price: {quote['h']}")
print(f"Lowest Price: {quote['l']}")
print(f"Percentage Change: {quote['chp']}%")
else:
print("Request Failed:", data["msg"])
This interface returns a clear JSON data structure that is easy to parse. In quantitative analysis, you can use the latest price to calculate real-time returns.
Step 2: Obtaining Historical Candlestick Data (Kline)
Historical candlestick data is the core of quantitative backtesting, supporting intervals from minute-level to monthly-level. Interface path: GET /stock/kline?region={region}&code={code}&kType={kType}&limit={limit}
For example, obtaining the last 100 daily candlesticks for Adidas:
import requests
import pandas as pd
from datetime import datetime
def fetch_historical_data(symbol, region="DE", kType=8, limit=100):
"""
Fetch Historical Candlestick Data
Parameters:
symbol: Stock code, e.g., "ADS"
region: Market code, "DE" for Germany
kType: Candlestick type, 1-minute, 2-5 minute, 8-daily, 9-weekly, 10-monthly
limit: Number of data bars to retrieve
"""
url = f"https://api.itick.org/stock/kline?region={region}&code={symbol}&kType={kType}&limit={limit}"
headers = {
"accept": "application/json",
"token": "your_token" # Replace with actual Token
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status() # Check if request is successful
data = response.json()
if data.get("code") == 0 and "data" in data:
# Convert data to Pandas DataFrame
df = pd.DataFrame(data["data"])
# Convert timestamp to readable format
df['datetime'] = pd.to_datetime(df['t'], unit='ms')
# Rename columns to standard financial data format
df.rename(columns={
'o': 'Open',
'h': 'High',
'l': 'Low',
'c': 'Close',
'v': 'Volume',
'tu': 'Turnover'
}, inplace=True)
# Select and sort columns
df = df[['datetime', 'Open', 'High', 'Low', 'Close', 'Volume', 'Turnover']]
df.set_index('datetime', inplace=True)
return df
else:
print(f"Data Fetch Failed: {data.get('msg')}")
return None
except requests.exceptions.RequestException as e:
print(f"Request Error: {e}")
return None
def analyze_german_stocks():
"""Analyze Historical Performance of Multiple German Stocks"""
symbols = ["ADS", "SAP", "VOW3", "ALV", "MRK"]
all_data = {}
for symbol in symbols:
print(f"Fetching historical data for {symbol}...")
df = fetch_historical_data(symbol, kType=8, limit=200) # Fetch 200 daily bars
if df is not None and len(df) > 0:
all_data[symbol] = df
# Calculate basic statistical indicators
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
# Calculate 20-day moving average
ma_20 = df['Close'].rolling(window=20).mean().iloc[-1]
print(f"{symbol}:")
print(f" Latest Close Price: {latest_close:.2f} EUR")
print(f" Daily Change: {daily_change:+.2f}%")
print(f" 20-Day Moving Average: {ma_20:.2f} EUR")
print(f" Data Time Range: {df.index[0].date()} to {df.index[-1].date()}")
print()
return all_data
if __name__ == "__main__":
# Fetch and analyze German stock data
stock_data = analyze_german_stocks()
# If data is fetched, further analysis can be performed
if stock_data:
print("Data fetching complete, ready for quantitative strategy backtesting and analysis!")
This helps identify trend reversal points.
Step 3: Obtaining Real-Time Order Book Depth (Depth)
The order book depth provides bid and ask data for five or ten levels, reflecting market order hanging situations. Interface path: 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"Stock Code: {depth['s']}")
print("Ask Side:")
for ask in depth['a'][:5]: # Display top 5 ask levels
print(f"Level {ask['po']}: Price {ask['p']}, Volume {ask['v']}, Orders {ask['o']}")
print("Bid Side:")
for bid in depth['b'][:5]: # Display top 5 bid levels
print(f"Level {bid['po']}: Price {bid['p']}, Volume {bid['v']}, Orders {bid['o']}")
else:
print("Request Failed:", data["msg"])
In quantitative analysis, order book data can be used to calculate bid-ask pressure ratios, helping to gauge market sentiment.
Step 4: Implementing Real-Time Push via WebSocket
For high-frequency quantitative analysis, RESTful APIs may have delays; WebSocket is recommended. After connection, subscribe to data supporting tick, quote, and depth types.
Python example (using websocket library):
import websocket
import json
import threading
import time
# WebSocket connection URL and Token
WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_token" # Replace with actual Token
def on_message(ws, message):
"""Handle received messages"""
data = json.loads(message)
# Handle successful connection message
if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
print("Connection successful, awaiting authentication...")
# Handle authentication result
elif data.get("resAc") == "auth":
if data.get("code") == 1:
print("Authentication successful")
subscribe(ws) # Subscribe after successful authentication
else:
print("Authentication failed")
ws.close()
# Handle subscription result
elif data.get("resAc") == "subscribe":
if data.get("code") == 1:
print("Subscription successful")
else:
print("Subscription failed:", data.get("msg"))
# Handle market data
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"Tick Data {symbol}: Latest Price={market_data['ld']}, Volume={market_data['v']}, Time={market_data['t']}")
elif data_type == "quote":
print(f"Quote Data {symbol}: Open={market_data['o']}, High={market_data['h']}, Low={market_data['l']}, Close={market_data['ld']}")
elif data_type == "depth":
print(f"Depth Data {symbol}: Bid1 Price={market_data['b'][0]['p'] if market_data['b'] else 'N/A'}, "
f"Ask1 Price={market_data['a'][0]['p'] if market_data['a'] else 'N/A'}")
def on_error(ws, error):
"""Handle errors"""
print("Error:", error)
def on_close(ws, close_status_code, close_msg):
"""Connection close callback"""
print("Connection closed")
def on_open(ws):
"""Callback after connection is established"""
print("WebSocket connection opened")
def subscribe(ws):
"""Subscribe to market data"""
subscribe_msg = {
"ac": "subscribe",
# Subscribe to real-time data for German Adidas, SAP, and Volkswagen
"params": "ADS$DE,SAP$DE,VOW3$DE",
"types": "tick,quote,depth" # Subscribe to tick, quote, and depth data
}
ws.send(json.dumps(subscribe_msg))
print("Subscription message sent")
def send_ping(ws):
"""Send heartbeat packets periodically to maintain connection"""
while True:
time.sleep(30) # Send heartbeat every 30 seconds
ping_msg = {
"ac": "ping",
"params": str(int(time.time() * 1000))
}
ws.send(json.dumps(ping_msg))
print("Heartbeat packet sent")
if __name__ == "__main__":
# Create WebSocket connection, pass Token via header
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
)
# Start heartbeat mechanism in a separate thread
ping_thread = threading.Thread(target=send_ping, args=(ws,))
ping_thread.daemon = True
ping_thread.start()
# Start WebSocket connection
ws.run_forever()
This code establishes a connection to the iTick WebSocket server and subscribes to real-time data for three well-known German companies (Adidas, SAP, and Volkswagen). After the connection is established, the server will continuously push three types of data:
- Tick Data: Includes the latest trade price, trade volume, and timestamp
- Quote Data: Includes open, high, low, close (OHLC) data such as opening price, highest price, lowest price, and latest price
- Depth Data: Includes bid and ask volumes and prices for each of the five levels
The advantages of obtaining real-time data via WebSocket lie in low latency and efficient data push mechanisms, particularly suitable for quantitative strategies that require real-time market monitoring and quick trading decisions.
Quantitative Analysis Example: Building a Simple Strategy
Obtaining data is just the first step; the real value lies in how to utilize this data for quantitative analysis. Below, we combine real-time and historical data to build a simple quantitative analysis example.
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
class GermanStockAnalyzer:
"""German Stock Analyzer"""
def __init__(self, historical_data):
self.data = historical_data
def calculate_technical_indicators(self):
"""Calculate Common Technical Indicators"""
df = self.data.copy()
# Calculate Moving Averages
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()
# Calculate Relative Strength Index (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))
# Calculate Bollinger Bands
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
# Calculate Volume Weighted Average Price (VWAP) - Intraday Indicator
df['VWAP'] = (df['Turnover'] / df['Volume']).rolling(window=20).mean()
return df
def generate_signals(self, df):
"""Generate Trading Signals Based on Technical Indicators"""
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
signals['signal'] = 0
# Dual Moving Average Crossover Strategy
signals['ma_signal'] = 0
signals.loc[df['MA_5'] > df['MA_20'], 'ma_signal'] = 1 # Golden Cross
signals.loc[df['MA_5'] < df['MA_20'], 'ma_signal'] = -1 # Death Cross
# RSI Overbought/Oversold Signals
signals['rsi_signal'] = 0
signals.loc[df['RSI'] < 30, 'rsi_signal'] = 1 # Oversold, Buy Signal
signals.loc[df['RSI'] > 70, 'rsi_signal'] = -1 # Overbought, Sell Signal
# Bollinger Bands Breakout Signals
signals['bb_signal'] = 0
signals.loc[df['Close'] < df['BB_lower'], 'bb_signal'] = 1 # Break Below Lower Band, Buy Signal
signals.loc[df['Close'] > df['BB_upper'], 'bb_signal'] = -1 # Break Above Upper Band, Sell Signal
# Combined Signal
signals['combined_signal'] = signals[['ma_signal', 'rsi_signal', 'bb_signal']].mean(axis=1)
# Generate Final Trading Signals
signals.loc[signals['combined_signal'] > 0.3, 'signal'] = 1 # Strong Buy
signals.loc[signals['combined_signal'] < -0.3, 'signal'] = -1 # Strong Sell
return signals
def plot_analysis(self, df, signals):
"""Visualize Analysis Results"""
fig, axes = plt.subplots(3, 1, figsize=(15, 12))
# Price and Moving Averages
ax1 = axes[0]
ax1.plot(df.index, df['Close'], label='Close Price', linewidth=1)
ax1.plot(df.index, df['MA_5'], label='5-Day MA', linewidth=1, alpha=0.7)
ax1.plot(df.index, df['MA_20'], label='20-Day MA', linewidth=1, alpha=0.7)
ax1.plot(df.index, df['MA_60'], label='60-Day MA', linewidth=1, alpha=0.7)
# Mark Trading Signals
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='Buy Signal')
ax1.scatter(sell_signals.index, df.loc[sell_signals.index, 'Close'],
color='red', marker='v', s=100, label='Sell Signal')
ax1.set_title('German Stock Price and Moving Averages')
ax1.set_ylabel('Price (EUR)')
ax1.legend()
ax1.grid(True, alpha=0.3)
# RSI Indicator
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='Overbought Line')
ax2.axhline(y=30, color='green', linestyle='--', alpha=0.5, label='Oversold Line')
ax2.fill_between(df.index, 30, 70, alpha=0.1, color='gray')
ax2.set_title('Relative Strength Index (RSI)')
ax2.set_ylabel('RSI Value')
ax2.legend()
ax2.grid(True, alpha=0.3)
# Bollinger Bands
ax3 = axes[2]
ax3.plot(df.index, df['Close'], label='Close Price', linewidth=1)
ax3.plot(df.index, df['BB_middle'], label='Middle Band', linewidth=1, alpha=0.7)
ax3.plot(df.index, df['BB_upper'], label='Upper Band', linewidth=1, alpha=0.7, linestyle='--')
ax3.plot(df.index, df['BB_lower'], label='Lower Band', linewidth=1, alpha=0.7, linestyle='--')
ax3.fill_between(df.index, df['BB_lower'], df['BB_upper'], alpha=0.1)
ax3.set_title('Bollinger Bands')
ax3.set_ylabel('Price (EUR)')
ax3.set_xlabel('Date')
ax3.legend()
ax3.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
def backtest_strategy(self, signals, initial_capital=10000):
"""Simple Strategy Backtest"""
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: # Buy signal, no current position
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: # Sell signal, with current position
capital = position * current_price
position = 0
trades.append({
'date': signals.index[i],
'action': 'SELL',
'price': current_price,
'capital': capital
})
# Calculate final returns
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
}
# Usage Example
if __name__ == "__main__":
# Assume we have fetched historical data
# Use simulated data for demonstration here
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)
# Create analyzer instance
analyzer = GermanStockAnalyzer(historical_data)
# Calculate technical indicators
df_with_indicators = analyzer.calculate_technical_indicators()
# Generate trading signals
signals = analyzer.generate_signals(df_with_indicators)
# Visualize analysis
analyzer.plot_analysis(df_with_indicators, signals)
# Backtest strategy
backtest_result = analyzer.backtest_strategy(signals)
print("Strategy Backtest Results:")
print(f"Initial Capital: {backtest_result['initial_capital']:.2f} EUR")
print(f"Final Capital: {backtest_result['final_capital']:.2f} EUR")
print(f"Total Return: {backtest_result['total_return']:.2f}%")
print(f"Number of Trades: {len(backtest_result['trades'])}")
This quantitative analysis example demonstrates how to apply data obtained from the iTick API to practical quantitative strategies. By calculating technical indicators, generating trading signals, and conducting strategy backtesting, we can systematically evaluate the effectiveness of trading strategies.
API Integration and Quantitative Analysis Considerations
- Rate Limits and Subscriptions: APIs have call limits; monitor them in production environments.
- Data Accuracy: After obtaining data, perform completeness and accuracy checks, such as detecting missing values or abnormal prices (e.g., 0 or prices far beyond normal ranges). Use pandas methods like dropna() and replace() to handle dirty data.
- Real-Time Optimization: For high-frequency quantitative strategies, select API providers with local Frankfurt deployments (such as iTick) to reduce network latency; also set up reasonable data caching to minimize repeated requests.
- Extensions: iTick supports more markets and can be extended to multi-asset strategies.
Summary
By integrating the Frankfurt Stock Exchange API for real-time stock quotes and historical data via Python, we have built the core data pipeline for quantitative analysis. This is not only a technical implementation but also the beginning of data-driven decision-making—a stable and reliable data flow makes strategy backtesting more accurate and signal generation more timely, laying a solid foundation for exploring alpha opportunities in the rigorous European market. Now, you possess the capability to connect to important global financial markets; it's time to transform this data into your strategic advantage.
Warm Reminder: This article is for reference only and does not constitute any investment advice. Markets involve risks; invest with caution.
Reference Document: https://itick.org/blog/stock-api/free-german-stock-api-comparison
GitHub: https://github.com/itick-org/
Top comments (0)