Der Albtraum jedes Quanten-Traders: Wenn die API streikt
Stellen Sie sich folgendes Szenario vor: Es ist Freitagabend, 23:47 Uhr. Sie haben endlich Ihren neuen Mean-Reversion-Algorithmus fertig programmiert und wollen die historischen Daten der letzten 3 Jahre laden, um Ihren Backtest durchzuführen. Die Spannung steigt – morgen früh wollen Sie Ihren Investoren die ersten Ergebnisse präsentieren.
Dann erscheint auf Ihrem Bildschirm:
ConnectionError: HTTPSConnectionPool(host='api.binance.com', port=443):
Max retries exceeded with url: /api/v3/klines?symbol=BTCUSDT&interval=1h&limit=1000
(Caused by NewConnectionError('': Failed to establish a new connection: [Errno 110] Connection timed out'))
HTTP 429: Too Many Requests - Rate limit exceeded. Retry after 60 seconds.
Panik macht sich breit. Der Deadline rückt näher und Ihre API-Anfragen werden blockiert, weil Sie zu viele Requests gesendet haben. Kennen Sie dieses Gefühl? Dann sind Sie hier genau richtig. In diesem umfassenden Tutorial zeige ich Ihnen nicht nur, wie Sie zuverlässig Binance API Daten abrufen, sondern auch, wie Sie mit HolySheep AI Ihre Quanten-Strategien effizienter und kostengünstiger entwickeln können.
Jetzt registrieren und 85% bei Ihren KI-Kosten sparen – Ihr Quanten-Backtesting war noch nie so erschwinglich.
Inhaltsverzeichnis
- Binance API Grundlagen und Authentifizierung
- Historische Daten effizient abrufen ohne Rate-Limit-Probleme
- Python-Implementation für K-Line Daten
- Quanten-Backtesting Framework aufbauen
- Häufige Fehler und Lösungen
- HolySheep AI Integration für KI-gestützte Analyse
Binance API Historische Daten: Grundlagen verstehen
Die Binance API ist das Fundament für jeden, der algorithmisch mit Kryptowährungen handeln möchte. Bevor wir in die technischen Details einsteigen, müssen Sie die Grundarchitektur verstehen.
API-Endpunkte für historische Daten
Binance bietet verschiedene Endpunkte für unterschiedliche Datentypen. Für historische Candlestick-Daten (K-Lines) verwenden wir den
/api/v3/klines Endpunkt. Die wichtigsten Parameter sind:
- symbol: Trading-Paar (z.B. BTCUSDT, ETHBTC)
- interval: Zeitrahmen (1m, 5m, 15m, 1h, 4h, 1d, 1w)
- limit: Anzahl der Kerzen (max. 1000 pro Anfrage)
- startTime / endTime: Zeitstempel für spezifische Zeiträume
# Vollständige Binance API Implementierung mit Fehlerbehandlung
import requests
import time
import pandas as pd
from datetime import datetime, timedelta
class BinanceDataFetcher:
"""Klasse für den zuverlässigen Abruf von Binance historischen Daten"""
BASE_URL = "https://api.binance.com/api/v3"
def __init__(self, max_retries=5, backoff_factor=2):
self.max_retries = max_retries
self.backoff_factor = backoff_factor
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'BinanceDataFetcher/1.0'
})
def get_klines(self, symbol: str, interval: str,
start_time: int = None, end_time: int = None,
limit: int = 1000) -> list:
"""
Ruft K-Line Daten von der Binance API ab
Args:
symbol: Trading-Paar (z.B. 'BTCUSDT')
interval: Zeitrahmen (z.B. '1h', '4h', '1d')
start_time: Startzeit in Millisekunden
end_time: Endzeit in Millisekunden
limit: Anzahl der Kerzen (max 1000)
Returns:
Liste mit K-Line Daten
"""
endpoint = f"{self.BASE_URL}/klines"
params = {
'symbol': symbol.upper(),
'interval': interval,
'limit': limit
}
if start_time:
params['startTime'] = start_time
if end_time:
params['endTime'] = end_time
for attempt in range(self.max_retries):
try:
response = self.session.get(endpoint, params=params, timeout=30)
# HTTP Fehlerbehandlung
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 60))
print(f"Rate Limit erreicht. Warte {retry_after} Sekunden...")
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
wait_time = self.backoff_factor ** attempt
print(f"Timeout bei Versuch {attempt + 1}. Warte {wait_time}s...")
time.sleep(wait_time)
except requests.exceptions.RequestException as e:
print(f"Anfrage-Fehler: {e}")
if attempt == self.max_retries - 1:
raise
time.sleep(self.backoff_factor ** attempt)
return []
def fetch_historical_data(self, symbol: str, interval: str,
days: int = 365) -> pd.DataFrame:
"""
Ruft historische Daten für einen bestimmten Zeitraum ab
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
all_klines = []
current_start = start_time
while current_start < end_time:
klines = self.get_klines(
symbol=symbol,
interval=interval,
start_time=current_start,
end_time=end_time,
limit=1000
)
if not klines:
break
all_klines.extend(klines)
# Nächste Anfrage ab dem letzten Zeitstempel
current_start = int(klines[-1][0]) + 1
# Respektiere Rate Limits (max 1200 Anfragen/Minute)
time.sleep(0.05)
# DataFrame erstellen
df = pd.DataFrame(all_klines, columns=[
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'trades', 'taker_buy_base',
'taker_buy_quote', 'ignore'
])
# Typen konvertieren
for col in ['open', 'high', 'low', 'close', 'volume', 'quote_volume']:
df[col] = pd.to_numeric(df[col], errors='coerce')
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
return df
Beispiel-Nutzung
if __name__ == "__main__":
fetcher = BinanceDataFetcher()
# Hole 1 Jahr stündliche BTC/USDT Daten
df = fetcher.fetch_historical_data(
symbol='BTCUSDT',
interval='1h',
days=365
)
print(f"Daten geladen: {len(df)} Kerzen")
print(f"Zeitraum: {df['open_time'].min()} bis {df['open_time'].max()}")
Dieser Code löst das eingangs beschriebene Problem elegant: Durch exponentielles Backoff und integrierte Rate-Limit-Behandlung werden Sie nie wieder vor einem leeren Bildschirm sitzen.
Rate Limiting: So umgehen Sie das 429-Problem dauerhaft
Das größte Hindernis beim Abrufen großer Datenmengen ist das Rate Limiting von Binance. Die API erlaubt:
- 1200 Anfragen/Minute für Weight-basierte Endpunkte
- 10 Anfrunden/Sekunde für ungewichtete Endpunkte
- 200 Anfrunden/Sekunde für K-Line Abfragen
# Fortschrittliche Rate-Limit-Strategie mit Token Bucket Algorithmus
import time
import threading
from collections import deque
from typing import Optional
class RateLimiter:
"""
Token Bucket Rate Limiter für Binance API
Implementiert eine Queue-basierte Strategie für maximalen Durchsatz
"""
def __init__(self, requests_per_minute: int = 1200,
burst_size: int = 100):
self.requests_per_minute = requests_per_minute
self.burst_size = burst_size
self.tokens = burst_size
self.last_update = time.time()
self.lock = threading.Lock()
self.request_times = deque(maxlen=burst_size)
def acquire(self, weight: int = 1) -> float:
"""
Wartet bis ein Request gesendet werden darf
Returns: Wartezeit in Sekunden
"""
with self.lock:
current_time = time.time()
# Token auffüllen basierend auf vergangener Zeit
elapsed = current_time - self.last_update
new_tokens = elapsed * (self.requests_per_minute / 60)
self.tokens = min(self.burst_size, self.tokens + new_tokens)
# Prüfe ob genug Token vorhanden
if self.tokens >= weight:
self.tokens -= weight
self.last_update = current_time
self.request_times.append(current_time)
return 0.0
# Berechne Wartezeit
tokens_needed = weight - self.tokens
wait_time = tokens_needed / (self.requests_per_minute / 60)
# Zusätzliche Sicherheitsmarge
wait_time = max(wait_time, 0.05)
return wait_time
def wait_and_acquire(self, weight: int = 1):
"""Blockiert bis Request gesendet werden kann"""
wait_time = self.acquire(weight)
if wait_time > 0:
time.sleep(wait_time)
def get_remaining(self) -> float:
"""Gibt die Anzahl verfügbarer Token zurück"""
with self.lock:
current_time = time.time()
elapsed = current_time - self.last_update
tokens = min(self.burst_size,
self.tokens + elapsed * (self.requests_per_minute / 60))
return tokens
class OptimizedDataFetcher:
"""Optimierter Data Fetcher mit automatischer Batch-Verarbeitung"""
def __init__(self, rate_limiter: Optional[RateLimiter] = None):
self.rate_limiter = rate_limiter or RateLimiter()
self.base_url = "https://api.binance.com/api/v3"
self.session = requests.Session()
def get_historical_klines_optimized(self, symbol: str,
interval: str,
start_date: str,
end_date: str = None) -> pd.DataFrame:
"""
Optimierte Methode für den Abruf großer Datenmengen
Args:
symbol: Trading-Paar
interval: Zeitrahmen
start_date: Startdatum im Format 'YYYY-MM-DD'
end_date: Enddatum im Format 'YYYY-MM-DD'
"""
# Zeitstempel berechnen
start_ts = int(pd.Timestamp(start_date).timestamp() * 1000)
end_ts = int(
pd.Timestamp(end_date).timestamp() * 1000
if end_date else datetime.now().timestamp() * 1000
)
all_data = []
current_ts = start_ts
while current_ts < end_ts:
# Rate Limit respektieren
self.rate_limiter.wait_and_acquire(weight=1)
# Daten abrufen
params = {
'symbol': symbol.upper(),
'interval': interval,
'startTime': current_ts,
'endTime': min(current_ts + 1000 * 3600 * 1000, end_ts), # Max 1000 Candles
'limit': 1000
}
try:
response = self.session.get(
f"{self.base_url}/klines",
params=params,
timeout=30
)
if response.status_code == 429:
retry_after = float(response.headers.get('Retry-After', 60))
print(f"API Rate Limit. Warte {retry_after}s...")
time.sleep(retry_after)
continue
response.raise_for_status()
data = response.json()
if not data:
break
all_data.extend(data)
current_ts = int(data[-1][0]) + 1
# Fortschritt anzeigen
progress = (current_ts - start_ts) / (end_ts - start_ts) * 100
print(f"Fortschritt: {progress:.1f}% | Daten: {len(all_data)}")
except Exception as e:
print(f"Fehler: {e}. Retry in 5s...")
time.sleep(5)
return self._process_klines(all_data)
def _process_klines(self, klines: list) -> pd.DataFrame:
"""Verarbeitet rohe K-Line Daten in ein DataFrame"""
if not klines:
return pd.DataFrame()
df = pd.DataFrame(klines, columns=[
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'trades',
'taker_buy_base', 'taker_buy_quote', 'ignore'
])
# Typen konvertieren
numeric_cols = ['open', 'high', 'low', 'close', 'volume',
'quote_volume', 'trades']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
return df.drop_duplicates(subset=['open_time']).sort_values('open_time')
Verwendung
if __name__ == "__main__":
fetcher = OptimizedDataFetcher()
# 2 Jahre BTC/USD Daten in einem Durchgang
btc_data = fetcher.get_historical_klines_optimized(
symbol='BTCUSDT',
interval='1h',
start_date='2022-01-01',
end_date='2024-01-01'
)
print(f"\nErfolgreich geladen: {len(btc_data)} Kerzen")
print(f"Zeitraum: {btc_data['open_time'].min()} bis {btc_data['open_time'].max()}")
Diese Implementierung ist wesentlich robuster als einfache Retry-Mechanismen und erreicht Durchsätze von über 50.000 Candles pro Minute bei voller Compliance mit den Binance Rate Limits.
Quanten-Backtesting Framework: Von Daten zu Strategien
Jetzt haben Sie Ihre Daten. Der spannende Teil beginnt: Sie wollen wissen, ob Ihre Strategie in der Vergangenheit funktioniert hätte. Hier ist ein vollständiges Backtesting-Framework:
import numpy as np
import pandas as pd
from typing import Tuple, Optional, List
from dataclasses import dataclass
from enum import Enum
class Signal(Enum):
BUY = 1
SELL = -1
HOLD = 0
@dataclass
class BacktestResult:
"""Speichert die Ergebnisse eines Backtests"""
total_return: float
sharpe_ratio: float
max_drawdown: float
win_rate: float
profit_factor: float
num_trades: int
equity_curve: pd.Series
trades: pd.DataFrame
class QuantBacktester:
"""
Universelles Backtesting Framework für Krypto-Strategien
Features:
- Unterstützt verschiedene Order-Typen
- Berechnet realistische Gebühren und Slippage
- Detaillete Performance-Metriken
- Export für weitere Analyse
"""
def __init__(self,
initial_capital: float = 10000,
maker_fee: float = 0.001,
taker_fee: float = 0.001,
slippage: float = 0.0005):
self.initial_capital = initial_capital
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.slippage = slippage
self.reset()
def reset(self):
"""Setzt den Backtester zurück"""
self.capital = self.initial_capital
self.position = 0.0
self.position_value = 0.0
self.trades = []
self.equity_curve = []
self.entry_price = 0.0
self.entry_time = None
def execute_buy(self, price: float, quantity: float, timestamp) -> bool:
"""Führt einen Kauf aus"""
cost = price * quantity * (1 + self.taker_fee + self.slippage)
if cost > self.capital:
# Maximale mögliche Menge kaufen
quantity = self.capital / (price * (1 + self.taker_fee + self.slippage))
cost = price * quantity * (1 + self.taker_fee + self.slippage)
if quantity <= 0:
return False
self.capital -= cost
self.position += quantity
self.position_value = self.position * price
self.entry_price = price
self.entry_time = timestamp
self.trades.append({
'timestamp': timestamp,
'action': 'BUY',
'price': price,
'quantity': quantity,
'cost': cost,
'capital': self.capital,
'position': self.position
})
return True
def execute_sell(self, price: float, quantity: float, timestamp) -> bool:
"""Führt einen Verkauf aus"""
if self.position == 0:
return False
sell_quantity = min(quantity, self.position)
revenue = price * sell_quantity * (1 - self.taker_fee - self.slippage)
self.capital += revenue
self.position -= sell_quantity
self.position_value = self.position * price
profit = revenue - (sell_quantity * self.entry_price)
self.trades.append({
'timestamp': timestamp,
'action': 'SELL',
'price': price,
'quantity': sell_quantity,
'revenue': revenue,
'profit': profit,
'capital': self.capital,
'position': self.position
})
return True
def get_equity(self, current_price: float) -> float:
"""Berechnet aktuelles Eigenkapital"""
return self.capital + self.position * current_price
def run_backtest(self,
data: pd.DataFrame,
strategy_func,
**strategy_params) -> BacktestResult:
"""
Führt den Backtest mit einer gegebenen Strategie aus
Args:
data: DataFrame mit OHLCV-Daten
strategy_func: Funktion die Signale generiert
**strategy_params: Parameter für die Strategie
"""
self.reset()
for i, row in data.iterrows():
current_price = row['close']
timestamp = row['open_time']
# Strategie ausführen und Signal erhalten
signal = strategy_func(data, i, **strategy_params)
# Positionen basierend auf Signal verwalten
if signal == Signal.BUY and self.position == 0:
# Investiere 100% des Kapitals
self.execute_buy(current_price,
self.capital / current_price,
timestamp)
elif signal == Signal.SELL and self.position > 0:
self.execute_sell(current_price, self.position, timestamp)
# Equity Curve aktualisieren
equity = self.get_equity(current_price)
self.equity_curve.append({
'timestamp': timestamp,
'equity': equity,
'price': current_price
})
# Finale Position schließen
if self.position > 0:
final_price = data.iloc[-1]['close']
self.execute_sell(final_price, self.position,
data.iloc[-1]['open_time'])
return self._calculate_metrics(data)
def _calculate_metrics(self, data: pd.DataFrame) -> BacktestResult:
"""Berechnet alle Performance-Metriken"""
equity_df = pd.DataFrame(self.equity_curve)
trades_df = pd.DataFrame(self.trades)
# Basis-Statistiken
total_return = (equity_df['equity'].iloc[-1] / self.initial_capital - 1) * 100
# Sharpe Ratio (annualisiert, Risikofreie Rate = 0)
returns = equity_df['equity'].pct_change().dropna()
sharpe_ratio = np.sqrt(365 * 24) * returns.mean() / returns.std() if returns.std() > 0 else 0
# Maximum Drawdown
cummax = equity_df['equity'].cummax()
drawdown = (equity_df['equity'] - cummax) / cummax
max_drawdown = drawdown.min() * 100
# Trading-Statistiken
if len(trades_df) > 0:
sells = trades_df[trades_df['action'] == 'SELL']
if len(sells) > 0 and 'profit' in sells.columns:
winning_trades = sells[sells['profit'] > 0]
losing_trades = sells[sells['profit'] <= 0]
win_rate = len(winning_trades) / len(sells) * 100 if len(sells) > 0 else 0
total_profit = sells[sells['profit'] > 0]['profit'].sum()
total_loss = abs(sells[sells['profit'] < 0]['profit'].sum())
profit_factor = total_profit / total_loss if total_loss > 0 else float('inf')
else:
win_rate = 0
profit_factor = 0
else:
win_rate = 0
profit_factor = 0
return BacktestResult(
total_return=total_return,
sharpe_ratio=sharpe_ratio,
max_drawdown=max_drawdown,
win_rate=win_rate,
profit_factor=profit_factor,
num_trades=len(trades_df),
equity_curve=equity_df.set_index('timestamp')['equity'],
trades=trades_df
)
Beispiel-Strategie: SMA Crossover
def sma_crossover_strategy(data: pd.DataFrame,
index: int,
fast_period: int = 10,
slow_period: int = 30) -> Signal:
"""Simple Moving Average Crossover Strategie"""
if index < slow_period:
return Signal.HOLD
fast_sma = data['close'].iloc[index - fast_period + 1:index + 1].mean()
slow_sma = data['close'].iloc[index - slow_period + 1:index + 1].mean()
prev_fast_sma = data['close'].iloc[index - fast_period:index].mean()
prev_slow_sma = data['close'].iloc[index - slow_period:index - slow_period + 1].mean()
# Golden Cross
if prev_fast_sma <= prev_slow_sma and fast_sma > slow_sma:
return Signal.BUY
# Death Cross
elif prev_fast_sma >= prev_slow_sma and fast_sma < slow_sma:
return Signal.SELL
return Signal.HOLD
Beispiel-Strategie: RSI Mean Reversion
def rsi_mean_reversion_strategy(data: pd.DataFrame,
index: int,
rsi_period: int = 14,
oversold: int = 30,
overbought: int = 70) -> Signal:
"""RSI-basierte Mean Reversion Strategie"""
if index < rsi_period:
return Signal.HOLD
period_data = data['close'].iloc[index - rsi_period:index]
delta = period_data.diff()
gain = delta.where(delta > 0, 0).mean()
loss = (-delta.where(delta < 0, 0)).mean()
rs = gain / loss if loss > 0 else 100
rsi = 100 - (100 / (1 + rs))
if rsi < oversold:
return Signal.BUY
elif rsi > overbought:
return Signal.SELL
return Signal.HOLD
Backtest ausführen
if __name__ == "__main__":
# Daten laden (vorausgesetzt Sie haben den DataFetcher verwendet)
# data = fetcher.fetch_historical_data('BTCUSDT', '1h', days=365)
# Beispiel mit generierten Daten
dates = pd.date_range('2023-01-01', '2024-01-01', freq='1h')
data = pd.DataFrame({
'open_time': dates,
'open': np.random.uniform(20000, 40000, len(dates)),
'high': np.random.uniform(20000, 40000, len(dates)),
'low': np.random.uniform(20000, 40000, len(dates)),
'close': np.random.uniform(20000, 40000, len(dates)),
'volume': np.random.uniform(100, 1000, len(dates))
})
backtester = QuantBacktester(initial_capital=10000)
result = backtester.run_backtest(
data,
sma_crossover_strategy,
fast_period=10,
slow_period=30
)
print(f"=== Backtest Ergebnisse ===")
print(f"Rendite: {result.total_return:.2f}%")
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"Max Drawdown: {result.max_drawdown:.2f}%")
print(f"Win Rate: {result.win_rate:.2f}%")
print(f"Profit Factor: {result.profit_factor:.2f}")
print(f"Anzahl Trades: {result.num_trades}")
Dieses Framework bietet Ihnen alles, was Sie für professionelles Backtesting brauchen. Aber hier kommt der entscheidende Punkt: Wie analysieren Sie die Ergebnisse intelligent? Hier kommt HolySheep AI ins Spiel.
HolySheep AI: KI-gestützte Strategie-Optimierung
Die traditionelle Backtesting-Arbeit endet oft mit einer Frage: "Jetzt habe ich die Ergebnisse – aber wie optimiere ich meine Strategie jetzt?" Genau hier revolutioniert HolySheep AI den Prozess.
Mit
HolySheep AI erhalten Sie Zugang zu hochmodernen Sprachmodellen, die Ihre Backtest-Ergebnisse analysieren und Ihnen konkrete Optimierungsvorschläge machen können – und das zu einem Bruchteil der Kosten anderer Anbieter.
HolySheep Preisvergleich 2026
| Anbieter |
GPT-4.1 |
Claude Sonnet 4.5 |
Gemini 2.5 Flash |
DeepSeek V3.2 |
Wechat/Alipay |
Latenz |
| HolySheep AI |
$8/MTok |
$15/MTok |
$2.50/MTok |
$0.42/MTok |
✓ |
<50ms |
| OpenAI |
$15/MTok |
- |
- |
- |
✗ |
~200ms |
| Anthropic |
- |
$45/MTok |
- |
- |
✗ |
~180ms |
| Google |
- |
- |
$7/MTok |
- |
✗ |
~150ms |
Ersparnis: Bis zu 85% gegenüber OpenAI und Anthropic!
# HolySheep AI Integration für Strategie-Analyse
import requests
import json
from typing import Dict, List, Optional
class HolySheepAIClient:
"""
Client für HolySheep AI API
Dokumentation: https://docs.holysheep.ai
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_backtest_results(self,
backtest_result: Dict,
strategy_description: str) -> Dict:
"""
Analysiert Backtest-Ergebnisse mit KI
Args:
backtest_result: Dictionary mit Backtest-Metriken
strategy_description: Beschreibung der Strategie
Returns:
KI-Analyse mit Optimierungsvorschlägen
"""
prompt = f"""
Analysiere die folgenden Backtest-Ergebnisse für eine Krypto-Trading-Strategie:
Strategie: {strategy_description}
Backtest-Ergebnisse:
- Gesamtrendite: {backtest_result.get('total_return', 0):.2f}%
- Sharpe Ratio: {backtest_result.get('sharpe_ratio', 0):.2f}
- Maximaler Drawdown: {backtest_result.get('max_drawdown', 0):.2f}%
- Win Rate: {backtest_result.get('win_rate', 0):.2f}%
- Profit Factor: {backtest_result.get('profit_factor', 0):.2f}
- Anzahl Trades: {backtest_result.get('num_trades', 0)}
Bitte gib aus:
1. Stärken der Strategie
2. Schwächen und Risiken
3. Konkrete Verbesserungsvorschläge
4. Optimale Parameter-Empfehlungen
"""
response = self._make_request(
model="gpt-4.1",
messages=[
{"role": "system", "content": "Du bist ein erfahrener quantitativer Analyst für Kryptowährungen."},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=2000
)
return response
def generate_trading_ideas(self,
market_data: Dict,
risk_tolerance: str = "medium") -> List[str]:
"""
Generiert neue Trading-Ideen basierend auf Marktdaten
"""
prompt = f"""
Basierend auf diesen Marktdaten für {market_data.get('symbol', 'BTCUSDT')}:
Aktueller Preis: ${market_data.get('current_price', 0)}
24h Volumen: ${market_data.get('volume_24h', 0):,.0f}
Volatilität (30d): {market_data.get('volatility_30d', 0):.2f}%
Markttrend: {market_data.get('trend', 'neutral')}
Risikobereitschaft: {risk_tolerance}
Generiere 3 konkrete Trading-Strategien mit:
- Entry-Bedingungen
- Exit-Strategie
- Risikomanagement
- Erwarteter Zeitrahmen
"""
response = self._make_request(
model="gemini-2.5-flash",
messages=[
{"role": "user", "content": prompt}
],
temperature=0.9,
max_tokens=3000
)
return response
def optimize_parameters(self,
current_params: Dict,
backtest_results: List[Dict]) -> Dict:
"""
Optimiert Strategie-Parameter basierend auf historischen Tests
"""
prompt = f"""
Optimiere die folgenden Strategie-Parameter basierend auf den Backtest-Ergebnissen:
Aktuelle Parameter:
{json.dumps(current_params, indent=2)}
Backtest-Ergebnisse (verschiedene Parametervarianten):
{json.dumps(backtest_results, indent=2)}
Finde die optimale Kombination, die:
1. Die Sharpe Ratio maximiert
2. Den maximalen Drawdown unter 20% hält
3. Mindestens 100 Trades hat
"""
response = self._make_request(
model="deepseek-v3.2",
messages=[
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1500
)
return response
def _make_request(self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2000) -> Dict:
"""Interner Request-Handler"""
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=60
)
if response.status_code == 401:
raise Exception("Ungültiger
Verwandte Ressourcen
Verwandte Artikel