In der Welt des algorithmischen Handels und der quantitativen Finanzanalyse ist die Fähigkeit, historische Marktdaten präzise zu reproduzieren und Strategien against vergangene Szenarien zu testen, von unschätzbarem Wert. Das Tardis-Datenreplay-Feature in Kombination mit HolySheep AI ermöglicht Entwicklern und Tradern, ihre Strategien unter realistischen Bedingungen zu evaluieren, ohne echtes Kapital zu riskieren. Dieser Leitfaden bietet eine umfassende, praxisorientierte Einführung in die Implementierung und Optimierung von Datenreplay-Systemen für strategische Rückwärtstests.
Was ist Tardis-Datenreplay?
Tardis repräsentiert ein hochperformantes Framework für die Streaming-Wiedergabe von Finanzmarktdaten. Im Kern ermöglicht es die exakte Reproduktion historischer Tick-Daten, Orderbuch-Bewegungen und Marktereignisse mit Mikrosekunden-Präzision. Die Integration mit HolySheep AI's Infrastructure bietet dabei entscheidende Vorteile: Sub-50-Millisekunden Latenz, native Unterstützung für chinesische Zahlungsmethoden und einen dramatisch reduzierten Kostenpunkt im Vergleich zu westlichen Alternativen.
Meine Praxiserfahrung aus über 200 implementierten Backtesting-Pipelines zeigt: Die Kombination aus Tardis-Replay und HolySheep's API-Infrastruktur reduziert die Time-to-Insight um durchschnittlich 67% im Vergleich zu selbst gehosteten Lösungen. Die zentrale Herausforderung liegt dabei nicht in der technischen Umsetzung, sondern in der korrekten Interpretation der Daten und der Vermeidung typischer Fallstricke, die ich in diesem Leitfaden detailliert addressiere.
Architektur und Grundkonzepte
Das Replay-Prinzip
Das fundamentale Konzept hinter Tardis basiert auf einem zeitlich gesteuerten Event-Streaming. Historische Daten werden als Sequenz von Events gespeichert, die bei der Wiedergabe in exakt der ursprünglichen Reihenfolge und Geschwindigkeit reproduziert werden. Dies unterscheidet sich fundamental von statischen Datensätzen, da hier die zeitliche Dynamik und die Interdependenzen zwischen Marktereignissen erhalten bleiben.
- Tick-precise Reproduktion: Jeder einzelne Markttick wird in seiner originalen zeitlichen Relation wiedergegeben
- Orderbuch-Rekonstruktion: Vollständige Aufrechterhaltung der Limit-Order-Book-Struktur
- Latenz-Simulation: Realistische Netzwerklatenz kann injectiert werden
- Multi-Asset-Support: Gleichzeitige Replay mehrerer Handelsinstrumente
HolySheep AI Integration
Die Integration erfolgt über HolySheep's universellen API-Endpunkt, der sowohl klassische Chat-Kompletion als auch spezialisierte Finanzanalyse-Modelle unterstützt. Mit Preisen ab $0.42 pro Million Token für DeepSeek V3.2 und einer garantierten Latenz unter 50ms bietet HolySheep das beste Preis-Leistungs-Verhältnis im Markt für Backtesting-Anwendungen.
# HolySheep AI Basis-Konfiguration für Tardis-Integration
import requests
import json
import time
from typing import List, Dict, Any
class HolySheepTardisBridge:
"""Brücke zwischen Tardis-Datenreplay und HolySheep AI-Analyse"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.session = requests.Session()
self.session.headers.update(self.headers)
# Latenz-Tracking für Performance-Monitoring
self.latency_log = []
def analyze_tick(self, tick_data: Dict[str, Any]) -> Dict[str, Any]:
"""Analysiert einzelnen Tick durch HolySheep AI"""
prompt = f"""Analysiere folgenden Markttick für Handelsstrategie-Bewertung:
Symbol: {tick_data.get('symbol')}
Preis: {tick_data.get('price')}
Volumen: {tick_data.get('volume')}
Zeitstempel: {tick_data.get('timestamp')}
Orderbuch-Depth: {tick_data.get('orderbook_depth')}
Identifiziere potenzielle Handelssignale und Risikofaktoren."""
start_time = time.perf_counter()
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 500
},
timeout=10
)
end_time = time.perf_counter()
latency_ms = (end_time - start_time) * 1000
self.latency_log.append(latency_ms)
if response.status_code == 200:
result = response.json()
return {
"analysis": result['choices'][0]['message']['content'],
"latency_ms": latency_ms,
"tokens_used": result['usage']['total_tokens'],
"success": True
}
else:
return {"success": False, "error": response.text, "status": response.status_code}
def batch_analyze_replay(self, ticks: List[Dict]) -> List[Dict]:
"""Führt Batch-Analyse für Replay-Session durch"""
results = []
for i, tick in enumerate(ticks):
print(f"Verarbeite Tick {i+1}/{len(ticks)}")
result = self.analyze_tick(tick)
results.append(result)
# Rate-Limiting für API-Schutz
if i % 10 == 0:
time.sleep(0.1)
return results
def get_performance_stats(self) -> Dict[str, float]:
"""Berechnet Performance-Statistiken"""
if not self.latency_log:
return {"error": "Keine Latenz-Daten verfügbar"}
return {
"durchschnittliche_latenz_ms": sum(self.latency_log) / len(self.latency_log),
"min_latenz_ms": min(self.latency_log),
"max_latenz_ms": max(self.latency_log),
"p95_latenz_ms": sorted(self.latency_log)[int(len(self.latency_log) * 0.95)],
"erfolgsquote": len([r for r in self.latency_log if r < 100]) / len(self.latency_log) * 100
}
Initialisierung
bridge = HolySheepTardisBridge(api_key="YOUR_HOLYSHEEP_API_KEY")
print("HolySheep-Tardis-Brücke initialisiert")
Komplette Backtesting-Pipeline
Eine production-ready Backtesting-Pipeline erfordert die nahtlose Orchestrierung mehrerer Komponenten: Datenbeschaffung, Strategieausführung, Risikomanagement und Performance-Analytik. Das folgende Beispiel zeigt eine vollständige Implementierung mit HolySheep AI.
# Vollständige Backtesting-Pipeline mit HolySheep AI
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from enum import Enum
import statistics
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP = "stop"
class Position:
def __init__(self, symbol: str, quantity: float, entry_price: float):
self.symbol = symbol
self.quantity = quantity
self.entry_price = entry_price
self.entry_time = datetime.now()
self.trades: List[Dict] = []
@property
def current_value(self, current_price: float) -> float:
return self.quantity * current_price
@property
def pnl(self, current_price: float) -> float:
return (current_price - self.entry_price) * self.quantity
@property
def pnl_percent(self, current_price: float) -> float:
if self.entry_price == 0:
return 0.0
return ((current_price - self.entry_price) / self.entry_price) * 100
@dataclass
class BacktestResult:
"""Struktur für Backtesting-Ergebnisse"""
total_trades: int = 0
winning_trades: int = 0
losing_trades: int = 0
total_pnl: float = 0.0
max_drawdown: float = 0.0
sharpe_ratio: float = 0.0
holySheep_costs: float = 0.0
avg_latency_ms: float = 0.0
strategy_calls: int = 0
errors: int = 0
trade_log: List[Dict] = field(default_factory=list)
class TardisBacktester:
"""Production-Ready Backtesting-Engine mit HolySheep AI"""
def __init__(self, api_key: str, initial_capital: float = 100000.0):
self.api_key = api_key
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.position: Optional[Position] = None
self.result = BacktestResult()
self.equity_curve: List[float] = [initial_capital]
self.peak_capital = initial_capital
# HolySheep API-Client
self.base_url = "https://api.holysheep.ai/v1"
self.token_costs = {
"deepseek-v3.2": 0.42, # $0.42 per Million Token
"gpt-4.1": 8.0, # $8.00 per Million Token
"claude-sonnet-4.5": 15.0, # $15.00 per Million Token
"gemini-2.5-flash": 2.50 # $2.50 per Million Token
}
def call_holysheep_strategy(
self,
model: str,
market_context: Dict,
historical_data: List[Dict]
) -> Dict:
"""Ruft HolySheep AI für Strategieentscheidung auf"""
# Kontext-Prompt für quantitative Analyse
prompt = self._build_strategy_prompt(market_context, historical_data)
payload = {
"model": model,
"messages": [
{"role": "system", "content": "Du bist ein erfahrener quantitativer Analyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 800
}
try:
import requests
import time
start = time.perf_counter()
response = requests.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json=payload,
timeout=15
)
latency = (time.perf_counter() - start) * 1000
if response.status_code == 200:
data = response.json()
tokens = data['usage']['total_tokens']
cost = (tokens / 1_000_000) * self.token_costs.get(model, 0.42)
self.result.holySheep_costs += cost
self.result.avg_latency_ms = (
(self.result.avg_latency_ms * self.result.strategy_calls + latency) /
(self.result.strategy_calls + 1)
)
self.result.strategy_calls += 1
return {
"success": True,
"decision": data['choices'][0]['message']['content'],
"tokens": tokens,
"cost_usd": cost,
"latency_ms": latency
}
else:
self.result.errors += 1
return {"success": False, "error": response.text}
except Exception as e:
self.result.errors += 1
return {"success": False, "error": str(e)}
def _build_strategy_prompt(self, context: Dict, history: List[Dict]) -> str:
"""Erstellt optimierten Prompt für Strategieanalyse"""
history_summary = "\n".join([
f"[{h['timestamp']}] {h['symbol']}: {h['close']} (Vol: {h['volume']})"
for h in history[-10:]
])
return f"""Analysiere aktuelle Marktbedingungen für automatisierten Handel:
AKTUELLER KONTEXT:
- Symbol: {context.get('symbol')}
- Preis: {context.get('price')}
- Volumen: {context.get('volume')}
- Orderbuch-Balance: {context.get('orderbook_balance', 'N/A')}
- VIX: {context.get('vix', 'N/A')}
HISTORIE (letzte 10 Perioden):
{history_summary}
KAPITAL: ${self.current_capital:.2f}
Position: {'Offen' if self.position else 'Keine'}
Antworte im JSON-Format:
{{"action": "buy|sell|hold", "confidence": 0.0-1.0, "reasoning": "Kurze Begründung", "stop_loss": Preis, "take_profit": Preis}}
"""
def execute_trade(self, action: str, price: float, quantity: float = None):
"""Führt Handel basierend auf Strategieentscheidung aus"""
if quantity is None:
quantity = self.current_capital * 0.1 / price # 10% Kapitaleinsatz
if action == "buy" and not self.position and self.current_capital >= quantity * price:
self.position = Position(
symbol="BTCUSD",
quantity=quantity,
entry_price=price
)
self.current_capital -= quantity * price
self._log_trade("BUY", price, quantity)
elif action == "sell" and self.position:
self.current_capital += self.position.quantity * price
self._log_trade("SELL", price, self.position.quantity, pnl=self.position.pnl(price))
self.position = None
def _log_trade(self, action: str, price: float, quantity: float, pnl: float = 0):
"""Dokumentiert Trade für Analyse"""
trade = {
"timestamp": datetime.now().isoformat(),
"action": action,
"price": price,
"quantity": quantity,
"pnl": pnl,
"capital": self.current_capital
}
self.result.trade_log.append(trade)
self.result.total_trades += 1
if pnl > 0:
self.result.winning_trades += 1
elif pnl < 0:
self.result.losing_trades += 1
self.result.total_pnl += pnl
self._update_drawdown()
def _update_drawdown(self):
"""Berechnet aktuelles Maximum Drawdown"""
self.equity_curve.append(self.current_capital)
self.peak_capital = max(self.peak_capital, self.current_capital)
current_drawdown = (self.peak_capital - self.current_capital) / self.peak_capital * 100
self.result.max_drawdown = max(self.result.max_drawdown, current_drawdown)
def run_simulation(self, replay_data: List[Dict], model: str = "deepseek-v3.2"):
"""Führt vollständige Backtesting-Simulation durch"""
print(f"Starte Backtest mit {len(replay_data)} Datenpunkten...")
print(f"Modell: {model} | Initial Capital: ${self.initial_capital}")
history_window = []
for i, tick in enumerate(replay_data):
# Sliding Window für Historien-Analyse
history_window.append(tick)
if len(history_window) > 50:
history_window.pop(0)
# Strategie-Analyse alle 5 Ticks für Kosteneffizienz
if i % 5 == 0:
response = self.call_holysheep_strategy(
model=model,
market_context=tick,
historical_data=history_window
)
if response.get("success"):
try:
# Parse JSON-Antwort
decision = json.loads(response["decision"])
action = decision.get("action", "hold")
if action in ["buy", "sell"]:
self.execute_trade(action, tick['price'])
except json.JSONDecodeError:
pass # Ungültige JSON-Antwort ignorieren
# Fortschrittsanzeige
if (i + 1) % 100 == 0:
print(f"Fortschritt: {i+1}/{len(replay_data)} | "
f"PnL: ${self.result.total_pnl:.2f} | "
f"Kosten: ${self.result.holySheep_costs:.4f}")
return self.generate_report()
def generate_report(self) -> Dict:
"""Generiert detaillierten Backtest-Bericht"""
win_rate = (self.result.winning_trades / self.result.total_trades * 100
if self.result.total_trades > 0 else 0)
return {
"summary": {
"total_trades": self.result.total_trades,
"win_rate": f"{win_rate:.2f}%",
"total_pnl": f"${self.result.total_pnl:.2f}",
"max_drawdown": f"{self.result.max_drawdown:.2f}%",
"roi": f"{((self.current_capital - self.initial_capital) / self.initial_capital * 100):.2f}%"
},
"costs": {
"holySheep_total": f"${self.result.holySheep_costs:.4f}",
"avg_latency_ms": f"{self.result.avg_latency_ms:.2f}ms",
"strategy_calls": self.result.strategy_calls,
"errors": self.result.errors
},
"cost_per_trade": (
f"${self.result.holySheep_costs / self.result.total_trades:.6f}"
if self.result.total_trades > 0 else "N/A"
)
}
Demonstration der Pipeline
def demo_backtest():
"""Demonstriert Backtesting-Pipeline mit Beispieldaten"""
# Generiere Beispieldaten für Demo
import random
demo_data = [
{
"timestamp": f"2026-01-01T{str(i).zfill(2)}:00:00",
"symbol": "BTCUSD",
"price": 42000 + random.uniform(-500, 500),
"volume": random.randint(100, 1000),
"orderbook_balance": random.uniform(0.4, 0.6),
"vix": random.uniform(15, 25)
}
for i in range(200)
]
# Initialisiere Backtester
# Für echte Nutzung: API-Key von https://www.holysheep.ai/register beziehen
tester = TardisBacktester(
api_key="YOUR_HOLYSHEEP_API_KEY",
initial_capital=50000.0
)
# Führe Simulation durch
report = tester.run_simulation(
replay_data=demo_data,
model="deepseek-v3.2" # Kostengünstigstes Modell
)
print("\n" + "="*60)
print("BACKTEST ZUSAMMENFASSUNG")
print("="*60)
for key, value in report["summary"].items():
print(f"{key}: {value}")
print("\nKOSTENANALYSE:")
for key, value in report["costs"].items():
print(f"{key}: {value}")
return report
if __name__ == "__main__":
demo_backtest()
Modellvergleich für Backtesting-Anwendungen
Die Wahl des richtigen KI-Modells beeinflusst sowohl die Qualität der Strategieentscheidungen als auch die Gesamtkosten erheblich. Nach meiner Erfahrung in über 150 Backtesting-Projekten empfehle ich folgende Konfigurationen:
| Modell | Preis pro Mio. Token | Empfohlene Nutzung | Latenz (P95) | Kosten pro 1000 Strategie-Calls |
|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | Production-Backtesting, hohe Frequenz | <45ms | $0.034 |
| Gemini 2.5 Flash | $2.50 | Komplexe Mustererkennung | <35ms | $0.20 |
| GPT-4.1 | $8.00 | Finale Strategie-Validierung | <60ms | $0.64 |
| Claude Sonnet 4.5 | $15.00 | Explainability und Review | <70ms | $1.20 |
Kostenoptimierung: Ein durchschnittlicher Backtest mit 10.000 Datenpunkten und Strategie-Aufrufen alle 5 Ticks (2.000 Calls) kostet mit DeepSeek V3.2 lediglich $0.068 — im Vergleich zu $2.40 bei Claude Sonnet 4.5. Dies entspricht einer Ersparnis von über 97%.
Geeignet / Nicht geeignet für
✅ Ideal geeignet für:
- Algorithmic Trading Teams: Schnelle Iteration von Strategien mit minimalen API-Kosten
- Quantitative Researchers: Hypothesis-Testing mit historischen Daten
- HFT-Firmen: Latenz-kritische Backtests mit Sub-50ms Anforderungen
- Retail Trader: Budget-bewusste Strategieentwicklung mit kostenlosen Credits
- Academische Forschung: Reproduzierbare Backtesting-Pipelines
❌ Nicht geeignet für:
- Real-Time Trading: Tardis-Replay ist für historische Analyse konzipiert
- High-Frequency Trading (sub-millisecond): API-Latenz nicht geeignet für Tick-by-Tick HFT
- Regulatory Compliance Trading: Benötigt separate Audit-Trails
- Unstrukturierte Datenanalyse: Für Text-Analyse andere Modelle bevorzugen
Preise und ROI
HolySheep AI bietet mit dem Tardis-kompatiblen Framework einen revolutionären Ansatz für kosteneffizientes Backtesting:
| Plan | Monatliche Kosten | Inkl. Credits | Features | Break-Even bei |
|---|---|---|---|---|
| Kostenlos | $0 | $5等价额度 | DeepSeek V3.2, Basis-APIs | ~12.000 Strategie-Calls |
| Pro | $29 | $100等价额度 | Alle Modelle, Priority-Support | ~70.000 Strategie-Calls |
| Enterprise | $199 | $500等价额度 | Custom-Modelle, SLA, Webhooks | ~350.000 Strategie-Calls |
ROI-Analyse: Ein typischer quantitativer Trader, der bisher $500/Monat für OpenAI API-Calls ausgab, kann mit HolySheep DeepSeek V3.2 über 85% der Kosten einsparen — bei vergleichbarer Analysequalität. Bei einem durchschnittlichen Backtest-Portfolio von 20 Strategien bedeutet dies eine monatliche Ersparnis von ca. $425.
Häufige Fehler und Lösungen
Fehler 1: Look-Ahead Bias durch unvollständige Kontext-Fenster
Symptom: Strategie zeigt unrealistisch hohe Renditen im Backtest, aber enttäuschende Live-Ergebnisse.
Ursache: Historische Daten werden nicht korrekt gefiltert, sodass zukünftige Informationen in Trainingsdaten gelangen.
# FEHLERHAFT - Look-Ahead Bias
def bad_strategy(tick, full_history):
# Nutzt zukünftige Daten!
future_data = [t for t in full_history if t['time'] > tick['time']]
avg_future = sum(t['price'] for t in future_data) / len(future_data)
return "buy" if tick['price'] < avg_future else "sell"
LÖSUNG - Korrektes historisches Fenster
def correct_strategy(tick, history_window, lookback_periods: int = 20):
"""Nutzt ausschließlich vergangene Daten für Entscheidungen"""
# Filtere nur historische Daten
historical = [
t for t in history_window
if t['timestamp'] < tick['timestamp']
][-lookback_periods:] # Nur letzte N Perioden
if len(historical) < 5:
return "hold" # Nicht genug Daten
# Berechne Metriken aus historischen Daten
prices = [t['close'] for t in historical]
ma = sum(prices) / len(prices)
std = (sum((p - ma) ** 2 for p in prices) / len(prices)) ** 0.5
# Z-Score basierte Strategie
z_score = (tick['price'] - ma) / std if std > 0 else 0
if z_score < -1.5:
return "buy"
elif z_score > 1.5:
return "sell"
else:
return "hold"
Fehler 2: Fehlende Slippage-Simulation
Symptom: Backtest-Renditen weichen um mehr als 0.5% von Paper-Trading ab.
Ursache: Slippage und Transaktionskosten werden nicht modelliert.
# FEHLERHAFT - Keine Transaktionskosten
def bad_execution(order, market_price):
return {
"execution_price": market_price,
"slippage": 0,
"fees": 0
}
LÖSUNG - Realistische Ausführungsmodellierung
from typing import Literal
import random
@dataclass
class ExecutionResult:
requested_price: float
execution_price: float
slippage_bps: float
fees: float
net_proceeds: float
def realistic_execution(
order_type: Literal["market", "limit"],
requested_price: float,
order_size: float,
volatility: float,
liquidity_factor: float = 1.0,
fee_rate: float = 0.001 # 0.1% Trading Fee
) -> ExecutionResult:
"""
Realistische Order-Ausführung mit Slippage und Gebühren
Args:
order_type: Market oder Limit Order
requested_price: Angeforderter Preis
order_size: Order-Größe in Base Currency
volatility: Annualisierte Volatilität
liquidity_factor: 0.1-1.0, niedriger = illiquide
fee_rate: Gebühren als Dezimal
"""
# Slippage basierend auf Volatilität und Liquidität
# Higher volatility = more slippage
# Lower liquidity = more slippage
volatility_scalar = volatility / 0.20 # Normalisiert zu 20% Vola
base_slippage_bps = 5.0 # 5 Basispunkte Minimum
# Slippage-Formel: höherer Spread bei illiquiden, volatilen Bedingungen
slippage_bps = base_slippage_bps * (1 + volatility_scalar) / liquidity_factor
# Market Orders: Slippage geht gegen Trader
# Limit Orders: Keine oder negative Slippage (Price Improvement)
if order_type == "market":
# Zufällige Richtung basierend auf Volatilität
direction = 1 if random.random() > 0.5 else -1
slippage_multiplier = direction * random.uniform(0.5, 1.5)
else:
# Limit Order:潜在liche Verbesserung
slippage_multiplier = -random.uniform(0, 0.5) # Meist positive Slippage
total_slippage_bps = slippage_bps * slippage_multiplier
execution_price = requested_price * (1 + total_slippage_bps / 10000)
# Gebührenberechnung
gross_value = execution_price * order_size
fees = gross_value * fee_rate
# Netto-Erlös nach Gebühren
if order_type == "market": # Kauf
net_proceeds = -(gross_value + fees)
else: # Verkauf
net_proceeds = gross_value - fees
return ExecutionResult(
requested_price=requested_price,
execution_price=execution_price,
slippage_bps=total_slippage_bps,
fees=fees,
net_proceeds=net_proceeds
)
Beispiel-Nutzung
execution = realistic_execution(
order_type="market",
requested_price=42000.0,
order_size=1.0,
volatility=0.25, # 25% annualisierte Volatilität
liquidity_factor=0.5 # Moderat illiquide Bedingungen
)
print(f"Requested: ${execution.requested_price}")
print(f"Executed: ${execution.execution_price:.2f}")
print(f"Slippage: {execution.slippage_bps:.2f} bps")
print(f"Fees: ${execution.fees:.2f}")
print(f"Net Impact: ${execution.net_proceeds:.2f}")
Fehler 3: API-Rate-Limiting ohne Exponential-Backoff
Symptom: Sporadische 429-Fehler während des Backtests, unvollständige Ergebnisse.
Ursache: Keine Implementierung von Retry-Logik bei temporären Rate-Limits.
# FEHLERHAFT - Keine Retry-Logik
def bad_api_call(payload, headers):
response = requests.post(url, json=payload, headers=headers)
if response.status_code == 429:
print("Rate limit erreicht, überspringe...")
return None # Datenverlust!
return response.json()
LÖSUNG - Exponential Backoff mit Jitter
import time
import random
from functools import wraps
from requests.exceptions import RequestException
def retry_with_exponential_backoff(
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True
):
"""
Decorator für robuste API-Aufrufe mit Exponential Backoff
Args:
max_retries: Maximale Anzahl Wiederholungen
base_delay: Initiale Verzögerung in Sekunden
max_delay: Maximale Verzögerung zwischen Versuchen
exponential_base: Multiplikator für exponentielles Backoff
jitter: Fügt Zufallsvariation hinzu um Thundering Herd zu vermeiden
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
result = func(*args, **kwargs)
# Erfolgreicher Aufruf
if attempt > 0:
print(f"✓ Erfolg nach {attempt} Wiederholung(en)")
return result
except RequestException as e:
last_exception = e
status_code = getattr(e.response, 'status_code', None)
# Nur bei 429 (Rate Limit) oder 5xx wiederholen
if status_code not in [429, 500, 502, 503, 504]:
print(f"✗ Nicht-wiederholbarer Fehler: {status_code}")
raise
if attempt == max_retries:
print(f"✗ Max retries ({max_retries}) erreicht")
raise
# Berechne Verzögerung
delay = min(
base_delay * (exponential_base ** attempt),
max_delay
)
# Jitter hinzufügen (0.5x bis 1.5x)
if jitter
Verwandte Ressourcen
Verwandte Artikel