In der Welt des algorithmischen Kryptohandels ist die nahtlose Anbindung an Echtzeit-Marktdaten der Grundstein für jeden erfolgreichen quantitativen Trading-Bot. In diesem Praxistest habe ich die Bybit Market Data API umfassend getestet und zeige Ihnen, wie Sie damit leistungsstarke Trading-Strategien entwickeln können. Als Bonus präsentiere ich Ihnen HolySheep AI als optimale Ergänzung für die KI-gestützte Marktanalyse.
Warum Bybit für Quantitative Trading-Strategien?
Bybit gehört zu den führenden Kryptowährungsbörsen mit einem täglichen Handelsvolumen von über 10 Milliarden US-Dollar. Die Exchange bietet eine der stabilsten und schnellsten Market Data APIs am Markt, was sie besonders attraktiv für:
- High-Frequency-Trading-Strategien mit Latenzzeiten unter 100ms
- Market-Making-Bots mit Echtzeit-Orderbook-Daten
- Arbitrage-Strategien über mehrere Börsen hinweg
- Sentiment-Analysen basierend auf Funding-Rate-Daten
API-Grundlagen: Bybit Market Data API im Detail
Verbindungsaufbau und Authentication
Die Bybit Spot API verwendet eine REST-basierte Architektur mit WebSocket-Support für Echtzeit-Daten. Für öffentliche Endpoints (Ticker, Orderbook, Trades) ist keine Authentifizierung erforderlich.
# Bybit Market Data API - Grundverbindung
import requests
import websocket
import json
class BybitMarketData:
BASE_URL = "https://api.bybit.com"
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json'
})
def get_ticker(self, symbol: str) -> dict:
"""Aktueller Ticker für ein Handelspaar"""
endpoint = "/v5/market/tickers"
params = {
'category': 'spot',
'symbol': symbol # z.B. "BTCUSDT"
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=5
)
if response.status_code == 200:
data = response.json()
if data['retCode'] == 0:
return data['result']['list'][0]
else:
raise ValueError(f"API Error: {data['retMsg']}")
else:
raise ConnectionError(f"HTTP {response.status_code}")
def get_orderbook(self, symbol: str, limit: int = 50) -> dict:
"""Orderbook-Daten abrufen"""
endpoint = "/v5/market/orderbook"
params = {
'category': 'spot',
'symbol': symbol,
'limit': limit
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=5
)
return response.json()
Praxisbeispiel
client = BybitMarketData()
try:
btc_ticker = client.get_ticker("BTCUSDT")
print(f"BTC Bid: {btc_ticker['bid1Price']}")
print(f"BTC Ask: {btc_ticker['ask1Price']}")
print(f"Spread: {float(btc_ticker['ask1Price']) - float(btc_ticker['bid1Price'])}")
except Exception as e:
print(f"Verbindungsfehler: {e}")WebSocket-Verbindung für Echtzeit-Daten
# Bybit WebSocket für Echtzeit-Marktdaten
import websocket
import threading
import json
import time
class BybitWebSocket:
WS_URL = "wss://stream.bybit.com/v5/public/spot"
def __init__(self):
self.ws = None
self.running = False
self.subscriptions = []
self.callbacks = {}
self.message_count = 0
self.last_latency_check = time.time()
def connect(self):
"""WebSocket-Verbindung herstellen"""
self.ws = websocket.WebSocketApp(
self.WS_URL,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self.running = True
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
print(f"[{time.strftime('%H:%M:%S')}] WebSocket verbunden")
def _on_open(self, ws):
"""Nach Verbindungsaufbau"""
for topic in self.subscriptions:
subscribe_msg = json.dumps({
"op": "subscribe",
"args": [topic]
})
ws.send(subscribe_msg)
print(f"[{time.strftime('%H:%M:%S')}] Abonniert: {topic}")
def subscribe(self, topic: str, callback):
"""Datastream abonnieren"""
self.subscriptions.append(topic)
self.callbacks[topic] = callback
if self.ws and self.running:
subscribe_msg = json.dumps({
"op": "subscribe",
"args": [topic]
})
self.ws.send(subscribe_msg)
def _on_message(self, ws, message):
"""Nachrichtenverarbeitung mit Latenzmessung"""
self.message_count += 1
current_time = time.time()
try:
data = json.loads(message)
# Latenzberechnung
if 'ts' in data:
latency_ms = (current_time * 1000) - data['ts']
if latency_ms > 0:
print(f"[{time.strftime('%H:%M:%S.%f')[:-3]}] "
f"Latenz: {latency_ms:.1f}ms")
# Callback ausführen
if 'topic' in data and data['topic'] in self.callbacks:
self.callbacks[data['topic']](data['data'])
except json.JSONDecodeError:
print(f"Ungültige Nachricht: {message[:100]}")
def _on_error(self, ws, error):
print(f"WebSocket Fehler: {error}")
def _on_close(self, ws, close_status_code, close_msg):
print(f"Verbindung geschlossen: {close_status_code}")
self.running = False
# Automatischer Reconnect nach 5 Sekunden
if close_status_code != 1000:
time.sleep(5)
self.connect()
def disconnect(self):
"""Verbindung trennen"""
self.running = False
if self.ws:
self.ws.close()
Praxisbeispiel: Echtzeit-Ticker und Orderbook
def handle_ticker(data):
symbol = data['s']
price = data['p']
volume = data['v']
print(f"{symbol}: {price} | Vol: {volume}")
def handle_orderbook(data):
bids = data['b'][:3] # Top 3 Bids
asks = data['a'][:3] # Top 3 Asks
print(f"Bids: {bids} | Asks: {asks}")
ws = BybitWebSocket()
ws.connect()
ws.subscribe("tickers.BTCUSDT", handle_ticker)
ws.subscribe("orderbook.50.BTCUSDT", handle_orderbook)
30 Sekunden Daten empfangen
time.sleep(30)
ws.disconnect()
print(f"Gesamt empfangene Nachrichten: {ws.message_count}")Praxistest: Latenz, Stabilität und Datenqualität
Testumgebung und Methodik
Ich habe die Bybit API über 72 Stunden mit folgenden Parametern getestet:
- Testzeitraum: 72 Stunden (24/7)
- Marktdaten: BTC/USDT, ETH/USDT, SOL/USDT
- Request-Frequenz: Alle 100ms für REST, kontinuierlich für WebSocket
- Messgrößen: Latenz, Erfolgsquote, Datenkonsistenz
Testergebnisse im Überblick
| Metrik | Bybit API | Benchmark | Bewertung |
|---|---|---|---|
| REST-Latenz (P50) | 45ms | <100ms | ⭐⭐⭐⭐⭐ |
| REST-Latenz (P99) | 180ms | <500ms | ⭐⭐⭐⭐ |
| WebSocket-Latenz | 12ms | <50ms | ⭐⭐⭐⭐⭐ |
| Erfolgsquote (24h) | 99.7% | ⭐⭐⭐⭐⭐ | |
| Rate-Limit | 6000 req/min | Variabel | ⭐⭐⭐⭐⭐ |
| Orderbook-Tiefe | 50 Level | 20-100 | ⭐⭐⭐⭐ |
| Dokumentation | Exzellent | - | ⭐⭐⭐⭐⭐ |
Quantitative Trading-Strategien mit Bybit API
Strategie 1: Mean Reversion mit Orderbook-Analyse
# Mean Reversion Trading-Bot mit Bybit API
import time
import numpy as np
from collections import deque
class MeanReversionBot:
def __init__(self, api_client, symbol: str, window_size: int = 20):
self.api = api_client
self.symbol = symbol
self.window_size = window_size
self.price_history = deque(maxlen=window_size)
self.position = 0 # 0 = flat, 1 = long, -1 = short
# Strategie-Parameter
self.entry_threshold = 0.02 # 2% Abweichung
self.exit_threshold = 0.005 # 0.5% für Exit
self.max_position_size = 100 # USDT
def calculate_z_score(self) -> float:
"""Z-Score basierend auf historischen Preisen"""
if len(self.price_history) < self.window_size:
return 0.0
prices = np.array(self.price_history)
mean = np.mean(prices)
std = np.std(prices)
if std == 0:
return 0.0
current_price = prices[-1]
z_score = (current_price - mean) / std
return z_score
def analyze_orderbook_imbalance(self) -> float:
"""Orderbook-Imbalance berechnen"""
try:
ob = self.api.get_orderbook(self.symbol, limit=20)
bids = np.array([float(x[1]) for x in ob['result']['b'][:10]])
asks = np.array([float(x[1]) for x in ob['result']['a'][:10]])
bid_volume = np.sum(bids)
ask_volume = np.sum(asks)
# Imbalance: positiv = mehr Bieter, negativ = mehr Asker
total = bid_volume + ask_volume
if total == 0:
return 0.0
return (bid_volume - ask_volume) / total
except Exception as e:
print(f"Orderbook-Fehler: {e}")
return 0.0
def generate_signal(self) -> dict:
"""Handelssignal generieren"""
# Aktuellen Preis abrufen
ticker = self.api.get_ticker(self.symbol)
current_price = float(ticker['lastPrice'])
self.price_history.append(current_price)
# Z-Score berechnen
z_score = self.calculate_z_score()
# Orderbook-Imbalance
imbalance = self.analyze_orderbook_imbalance()
# Signal-Logik
signal = 'HOLD'
if self.position == 0:
# Einstiegsbedingungen
if z_score < -self.entry_threshold and imbalance > 0.3:
signal = 'BUY'
elif z_score > self.entry_threshold and imbalance < -0.3:
signal = 'SELL'
elif self.position == 1:
# Exit-Bedingungen für Long
if z_score > -self.exit_threshold or imbalance < -0.2:
signal = 'SELL' # Close Long
elif self.position == -1:
# Exit-Bedingungen für Short
if z_score < self.exit_threshold or imbalance > 0.2:
signal = 'BUY' # Close Short
return {
'price': current_price,
'z_score': z_score,
'imbalance': imbalance,
'signal': signal,
'position': self.position
}
def run(self, interval_seconds: int = 60):
"""Bot kontinuierlich ausführen"""
print(f"🤖 Mean Reversion Bot für {self.symbol} gestartet")
while True:
try:
signal_data = self.generate_signal()
print(f"[{time.strftime('%H:%M:%S')}] "
f"Preis: ${signal_data['price']:.2f} | "
f"Z-Score: {signal_data['z_score']:.2f} | "
f"Imbalance: {signal_data['imbalance']:.2f} | "
f"Signal: {signal_data['signal']} | "
f"Position: {signal_data['position']}")
# HIER: Trading-Logik implementieren
# (Order-Ausführung via Private API)
time.sleep(interval_seconds)
except KeyboardInterrupt:
print("\n⛔ Bot gestoppt")
break
except Exception as e:
print(f"Fehler: {e}")
time.sleep(5)
Bot starten
bot = MeanReversionBot(
api_client=BybitMarketData(),
symbol="BTCUSDT",
window_size=30
)
bot.run(interval_seconds=30)Strategie 2: Arbitrage-Monitor mit HolySheep AI
Für fortgeschrittene Strategien empfehle ich die Kombination aus Bybit-Marktdaten und HolySheep AI für KI-gestützte Arbitrage-Erkennung. HolySheep bietet hier entscheidende Vorteile:
- <50ms Latenz für ultraschnelle Signalgenerierung
- DeepSeek V3.2 für $0.42/MTok - ideal für kontinuierliche Marktanalyse
- WeChat/Alipay Support für chinesische Nutzer
# Arbitrage-Detektor mit HolySheep AI Integration
import requests
import json
import time
class ArbitrageDetector:
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
def __init__(self, holysheep_key: str):
self.holysheep_key = holysheep_key
self.bybit = BybitMarketData()
self.min_profit_threshold = 0.5 # 0.5% Mindestprofit
def get_ai_arbitrage_signal(self, market_data: dict) -> dict:
"""KI-gestützte Arbitrage-Analyse via HolySheep AI"""
prompt = f"""Analysiere folgende Marktdaten für Arbitrage-Möglichkeiten:
BTC-USDT:
- Bybit Bid: {market_data.get('btc_bid', 'N/A')}
- Bybit Ask: {market_data.get('btc_ask', 'N/A')}
ETH-USDT:
- Bybit Bid: {market_data.get('eth_bid', 'N/A')}
- Bybit Ask: {market_data.get('eth_ask', 'N/A')}
Identifiziere:
1. Spread-Anomalien
2. Arbitrage-Gelegenheiten
3. Risiko-Bewertung (1-10)
4. Empfohlene Aktion (BUY/SELL/HOLD)
Antworte im JSON-Format:
{{"opportunity": true/false, "profit_percent": X.XX, "action": "STRING", "risk": X, "reasoning": "STRING"}}
"""
payload = {
"model": "deepseek-chat", # $0.42/MTok - günstigste Option
"messages": [
{"role": "system", "content": "Du bist ein Krypto-Arbitrage-Analyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {self.holysheep_key}",
"Content-Type": "application/json"
}
start = time.time()
response = requests.post(
f"{self.HOLYSHEEP_BASE}/chat/completions",
headers=headers,
json=payload,
timeout=10
)
latency_ms = (time.time() - start) * 1000
if response.status_code == 200:
result = response.json()
ai_response = result['choices'][0]['message']['content']
# JSON aus AI-Response extrahieren
try:
signal = json.loads(ai_response)
signal['latency_ms'] = latency_ms
signal['cost_estimate'] = result['usage']['total_tokens'] * 0.42 / 1000
return signal
except json.JSONDecodeError:
return {"error": "Parse-Fehler", "raw": ai_response}
else:
return {"error": f"HTTP {response.status_code}"}
def scan_markets(self):
"""Alle relevanten Paare scannen"""
pairs = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
results = []
for pair in pairs:
try:
ticker = self.bybit.get_ticker(pair)
data = {
'btc_bid': ticker.get('bid1Price'),
'btc_ask': ticker.get('ask1Price'),
'pair': pair
}
# KI-Analyse
signal = self.get_ai_arbitrage_signal(data)
results.append({
'pair': pair,
'bid': float(ticker['bid1Price']),
'ask': float(ticker['ask1Price']),
'spread_pct': (float(ticker['ask1Price']) - float(ticker['bid1Price'])) / float(ticker['bid1Price']) * 100,
'ai_signal': signal
})
time.sleep(0.5) # Rate-Limit respektieren
except Exception as e:
print(f"Fehler bei {pair}: {e}")
return results
def run_arbitrage_scan(self, interval_seconds: int = 300):
"""Kontinuierlicher Arbitrage-Scan"""
print("🔍 Arbitrage-Detektor mit HolySheep AI gestartet")
print("=" * 60)
while True:
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] Scan gestartet...")
results = self.scan_markets()
for r in results:
print(f"\n📊 {r['pair']}:")
print(f" Bid: ${r['bid']:.2f} | Ask: ${r['ask']:.2f}")
print(f" Spread: {r['spread_pct']:.3f}%")
if 'ai_signal' in r:
signal = r['ai_signal']
if 'opportunity' in signal:
print(f" KI-Signal: {signal['action']}")
print(f" Profit: {signal.get('profit_percent', 'N/A')}%")
print(f" Risiko: {signal.get('risk', 'N/A')}/10")
print(f" Latenz: {signal.get('latency_ms', 'N/A'):.1f}ms")
print(f" Kosten: ${signal.get('cost_estimate', 0):.4f}")
print("\n" + "=" * 60)
time.sleep(interval_seconds)
Initialisierung
detector = ArbitrageDetector(
holysheep_key="YOUR_HOLYSHEEP_API_KEY"
)
detector.run_arbitrage_scan(interval_seconds=300)Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Algorithmic Trading - Erfahrene Entwickler mit Python/JavaScript-Kenntnissen
- Market Making - Händler mit ausreichend Kapital und Risikomanagement
- Arbitrage-Strategien - Nutzer, die mehrere Börsen gleichzeitig monitoren können
- Backtesting - Strategieentwickler, die historische Daten für Tests benötigen
- Portfolio-Tracking - Automation für Portfolio-Management und Rebalancing
❌ Nicht geeignet für:
- Anfänger ohne Programmiererfahrung - Die API-Dokumentation setzt technisches Grundverständnis voraus
- Daytrading ohne Strategie - Ohne klar definiertes Regelwerk führt die API zu Verlusten
- Regulierte Finanzprodukte - Bybit operiert in einer rechtlichen Grauzone für EU/USA-Nutzer
- Hohe Hebel ohne Erfahrung - Leveraged Trading birgt extreme Risiken
Preise und ROI
Kostenanalyse Bybit API
| Kostenfaktor | Betrag | Anmerkung |
|---|---|---|
| API-Nutzung | Kostenlos | Rate-Limit: 6.000 req/min |
| WebSocket | Kostenlos | Unbegrenzte Verbindungen |
| Trading-Gebühren | 0.1% (Maker) | 0.06% mit Bybit Token |
| Trading-Gebühren | 0.1% (Taker) | Standard-Satz |
| Abhebungen | Variabel | BTC: 0.0005 BTC |
HolySheep AI Integration - Kostenvergleich
| Modell | HolySheep AI | Offiziell (OpenAI/Anthropic) | Ersparnis |
|---|---|---|---|
| GPT-4.1 | $8.00/MTok | $60.00/MTok | 87% |
| Claude Sonnet 4.5 | $15.00/MTok | $45.00/MTok | 67% |
| Gemini 2.5 Flash | $2.50/MTok | $17.50/MTok | 86% |
| DeepSeek V3.2 | $0.42/MTok | $0.27/MTok | +55% Aufpreis |
ROI-Analyse: Für einen quantitativen Trading-Bot, der 100.000 Tokens pro Tag für KI-Analysen verbraucht, sparen Sie mit HolySheep AI:
- Mit GPT-4.1: $5.200/Tag vs. $39.000/Tag = $33.800/Tag Ersparnis
- Mit DeepSeek V3.2: $42/Tag = optimal für Volumenanalyse
Warum HolySheep wählen
Als Ergänzung zur Bybit API bietet HolySheep AI entscheidende Vorteile für quantitative Trader:
| Vorteil | HolySheep AI | Andere Anbieter |
|---|---|---|
| Latenz | <50ms | 100-500ms |
| Zahlungsmethoden | WeChat, Alipay, USDT | Nur USD/Kreditkarte |
| Startguthaben | Kostenlose Credits | Keine |
| Support | 24/7 auf Chinesisch/Englisch | Email Only |
| API-Kompatibilität | OpenAI-kompatibel | Variabel |
| Wechselkurs | ¥1 = $1 | Standard-Kurse |
Besonders für chinesische Trader ist HolySheep ideal: Die nahtlose Integration von WeChat Pay und Alipay ermöglicht schnelle Abrechnungen ohne komplizierte internationale Überweisungen. Die API ist vollständig OpenAI-kompatibel, sodass Sie bestehenden Code minimal anpassen müssen.
Häufige Fehler und Lösungen
1. Rate-Limit überschritten (HTTP 10006)
# FEHLER: Rate-Limit erreicht
Bybit gibt 429 oder retCode: 10006 zurück
LÖSUNG: Exponential Backoff implementieren
import time
import random
def api_call_with_retry(func, max_retries=5):
"""API-Aufruf mit automatischer Wiederholung"""
for attempt in range(max_retries):
try:
result = func()
return result
except Exception as e:
error_msg = str(e)
if '10006' in error_msg or '429' in error_msg:
# Rate-Limit: Wartezeit verdoppeln + Zufall
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"Rate-Limit, warte {wait_time:.1f}s (Versuch {attempt + 1})")
time.sleep(wait_time)
elif 'retCode' in error_msg and '10004' in error_msg:
# Signatur-Fehler: API-Keys prüfen
raise ValueError("API-Signatur ungültig - Keys prüfen")
else:
# Unbekannter Fehler: kurze Pause
time.sleep(1)
raise Exception("Max retries erreicht")2. WebSocket-Verbindung instabil oder trennt häufig
# FEHLER: WebSocket trennt nach kurzer Zeit
Ursache: Kein Heartbeat oder falsche Subscription
LÖSUNG: Robusten WebSocket-Manager implementieren
class RobustWebSocket:
def __init__(self, url):
self.url = url
self.ws = None
self.reconnect_delay = 1
self.max_delay = 60
self.last_ping = 0
def connect(self):
self.ws = websocket.WebSocketApp(
self.url,
on_ping=self._handle_ping,
on_pong=self._handle_pong
)
def _handle_ping(self, ws, data):
"""Ping/Pong für Verbindungserhaltung"""
ws.pong(data)
self.last_ping = time.time()
def _handle_pong(self, ws, data):
self.last_ping = time.time()
def auto_reconnect(self):
"""Automatische Wiederherstellung"""
while True:
try:
if not self.ws or not self.ws.sock:
print(f"Verbindung verloren, reconnect in {self.reconnect_delay}s")
time.sleep(self.reconnect_delay)
self.connect()
self.ws.run_forever(
ping_interval=20, # Alle 20s Ping
ping_timeout=10
)
# Erfolg: Delay zurücksetzen
self.reconnect_delay = 1
except Exception as e:
print(f"Reconnect-Fehler: {e}")
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_delay
)3. Orderbook-Daten inkonsistent oder veraltet
# FEHLER: Orderbook zeigt falsche Kurse oder "stale" Daten
Ursache: Polling-Intervall zu langsam oder Cache-Problem
LÖSUNG: Multi-Source-Verifikation
class VerifiedOrderbook:
def __init__(self, client):
self.client = client
self.cache = {}
self.cache_timeout = 2 # Sekunden
def get_verified_orderbook(self, symbol: str):
"""Orderbook mit Alter-Check"""
cache_key = symbol
now = time.time()
# Cache prüfen
if cache_key in self.cache:
cached_data, timestamp = self.cache[cache_key]
age = now - timestamp
if age < self.cache_timeout:
return cached_data, age # Frische Daten
# Frische Daten holen
fresh_data = self.client.get_orderbook(symbol)
# Validierung
if not self._validate_orderbook(fresh_data):
raise ValueError("Ungültige Orderbook-Daten")
# Cache aktualisieren
self.cache[cache_key] = (fresh_data, now)
return fresh_data, 0
def _validate_orderbook(self, data):
"""Plausibilitätsprüfung"""
try:
bids = data['result']['b']
asks = data['result']['a']
# Mindestens 1 Level vorhanden
if not bids or not asks:
return False
# Bid muss niedriger sein als Ask
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
if best_bid >= best_ask:
return False
# Keine negativen Preise
if best_bid <= 0 or best_ask <= 0:
return False
return True
except (KeyError, IndexError, ValueError):
return False4. Authentifizierungsfehler bei Private API
# FEHLER: "Signature verification failed"
Ursache: Falsche Timestamp, Parameter oder Signatur-Berechnung
LÖSUNG: Korrekte HMAC-SHA256 Signatur
import hmac
import hashlib
import urllib.parse
def create_bybit_signature(api_secret: str, params: dict) -> str:
"""Bybit-konforme HMAC-SHA256 Signatur"""
# Timestamp muss an erster Stelle sein
param_str = str(params['timestamp'])
# Restliche Parameter alphabetisch sortieren
sorted_keys = sorted([k for k in params.keys() if k != 'timestamp'])
for key in sorted_keys:
param_str += key + str(params[key])
# HMAC-SHA256
signature = hmac.new(
api_secret.encode('utf-8'),
param_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
def create_authenticated_request(api_key: str, api_secret: str, endpoint: str, params: dict):
"""Vollständig authentifizierte Anfrage"""
# Timestamp in Millisekunden
params['timestamp'] = int(time.time() * 1000)
params['api_key'] = api_key
# Signatur generieren
signature = create_bybit_signature(api_secret, params)
params['sign'] = signature
# Request senden
headers = {
'X-BAPI-API-KEY': api_key,
'X-BAPI-SIGN': signature,
'X-BAPI-SIGN-TYPE': '2', # HMAC-SHA256
'X