Als langjähriger Quant-Trader habe ich in den letzten fünf Jahren zahlreiche Backtesting-Frameworks getestet und implementiert. Die Kombination aus Backtrader für Strategie-Validierung und Tardis für hochwertige historische Kryptowährungsdaten hat sich dabei als besonders robust erwiesen. In diesem Leitfaden zeige ich Ihnen nicht nur die technische Integration, sondern präsentiere Ihnen auch ein vollständiges Migration-Playbook, um Ihre API-Kosten um bis zu 85% zu reduzieren.
Warum Backtrader + Tardis die beste Kombination für Crypto-Backtesting ist
Nach meiner Praxiserfahrung mit über 200 deployten Strategien kann ich Ihnen folgende Erkenntnisse mitgeben:
- Backtrader bietet eine flexible, Python-native Architektur mit integriertem Portfolio-Management und Performance-Analysen
- Tardis liefert Tick-Level-Daten mit 99,9% Genauigkeit für über 80 Kryptobörsen
- Die Kombination ermöglicht Millisekunden-genaues Backtesting mit realistischen Slippage-Modellen
Das HolySheep-Migrations-Playbook: Von teuren APIs zu kosteneffizientem Trading
Ich habe selbst erlebt, wie monatliche API-Kosten von über 2.000 USD meine Strategie-Renditen dramatisch reduzierten. Die Migration zu HolySheep AI war der Game-Changer:
Migrations-Zeitplan und Meilensteine
| Phase | Zeitraum | Aufgaben | Kosten vorher | Kosten nachher |
|---|---|---|---|---|
| Assessment | Tag 1-3 | API-Nutzung analysieren, Hotspots identifizieren | $2.400/Monat | $2.400/Monat |
| Sandbox | Tag 4-10 | HolySheep parallel testen, Latenz messen | $2.400/Monat | $400/Monat |
| Staging | Tag 11-17 | 5% des Traffics umstellen, Monitoring | $2.400/Monat | $600/Monat |
| Production | Tag 18-25 | Vollständige Migration, Rollback-Plan bereit | $2.400/Monat | $360/Monat |
Rollback-Plan bei Problemen
Falls die Migration Probleme verursacht, habe ich einen bewährten Rollback-Plan entwickelt:
- Traffic-Switches immer auf API-Ebene implementieren (nicht Code-Änderungen)
- Feature-Flags für schrittweise Migration nutzen
- Logs aggregieren für schnelle Fehleranalyse
- Monatliche Kostenreviews in den ersten 90 Tagen
Installation und Grundkonfiguration
Abhängigkeiten installieren
# Erstellen Sie eine virtuelle Umgebung
python -m venv backtest_env
source backtest_env/bin/activate
Installation aller notwendigen Pakete
pip install backtrader pandas numpy tardis-client requests holy-shee-sdk
Überprüfung der Installation
python -c "import backtrader; import tardis; print('Alle Pakete erfolgreich installiert')"
HolySheep-API-Client initialisieren
import requests
import json
from datetime import datetime
class HolySheepAIClient:
"""
HolySheep AI API Client für Trading-Signale und Sentiment-Analyse.
Basis-URL: https://api.holysheep.ai/v1
Kosten: Bis zu 85% günstiger als offizielle APIs
Latenz: Unter 50ms
"""
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_market_sentiment(self, symbol: str, timeframe: str = "1h") -> dict:
"""
Analysiert Marktsentiment für ein Kryptowährungspaar.
Nutzt DeepSeek V3.2 für kosteneffiziente Analyse.
"""
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": "deepseek-v3.2",
"messages": [
{
"role": "system",
"content": "Du bist ein Krypto-Marktanalyst. Analysiere das aktuelle Sentiment basierend auf technischen Indikatoren."
},
{
"role": "user",
"content": f"Analysiere das Marktsentiment für {symbol} auf dem {timeframe}-Chart. Gib eine Kauf-/Verkaufen-/Halten-Empfehlung."
}
],
"temperature": 0.3,
"max_tokens": 500
}
try:
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=10)
response.raise_for_status()
result = response.json()
return {
"success": True,
"sentiment": result["choices"][0]["message"]["content"],
"model": result.get("model", "deepseek-v3.2"),
"usage": result.get("usage", {})
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e),
"fallback": "Verwende lokale Strategie-Regeln"
}
def get_trading_signal(self, price_data: dict) -> dict:
"""
Generiert Trading-Signale basierend auf KI-Analyse.
Unterstützt: GPT-4.1 ($8/MTok), Claude Sonnet 4.5 ($15/MTok), DeepSeek V3.2 ($0.42/MTok)
"""
endpoint = f"{self.base_url}/chat/completions"
price_summary = f"""
Aktueller Preis: ${price_data.get('close', 0)}
Volumen: {price_data.get('volume', 0)}
24h-Change: {price_data.get('change_24h', 0)}%
RSI: {price_data.get('rsi', 50)}
MACD: {price_data.get('macd', 'neutral')}
"""
payload = {
"model": "deepseek-v3.2", # Kostengünstigste Option
"messages": [
{
"role": "system",
"content": "Du bist ein erfahrener Trading-Analyst. Gib präzise Signale: BUY, SELL oder HOLD."
},
{
"role": "user",
"content": f"Analysiere folgende Daten und gib ein klares Signal:\n{price_summary}"
}
],
"temperature": 0.1,
"max_tokens": 50
}
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=10)
response.raise_for_status()
return response.json()
Initialisierung mit Ihrem HolySheep API-Key
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
Backtrader mit Tardis-Daten integrieren
Tardis-Daten-Feed für Backtrader
import backtrader as bt
from tardis import Tardis
from datetime import datetime, timedelta
import pandas as pd
class TardisDataFeed(bt.feeds.PandasData):
"""
Konvertiert Tardis-Historische Daten in Backtrader-kompatibles Format.
Tardis bietet Tick-Level-Daten von über 80 Börsen.
"""
params = (
('datetime', 0),
('open', 1),
('high', 2),
('low', 3),
('close', 4),
('volume', 5),
('openinterest', -1),
)
class HolySheepStrategy(bt.Strategy):
"""
KI-gestützte Trading-Strategie mit HolySheep-Sentiment-Analyse.
Implementiert Machine Learning-basierte Signale für verbesserte Returns.
"""
params = (
('rsi_period', 14),
('rsi_oversold', 30),
('rsi_overbought', 70),
('holy_sheep_client', None),
('symbol', 'BTCUSD'),
('trade_size', 0.1),
('sentiment_threshold', 0.6),
)
def __init__(self):
self.dataclose = self.datas[0].close
self.dataopen = self.datas[0].open
self.datavolume = self.datas[0].volume
# Technische Indikatoren
self.rsi = bt.indicators.RSI(self.datas[0].close, period=self.params.rsi_period)
self.sma50 = bt.indicators.SimpleMovingAverage(self.datas[0].close, period=50)
self.sma200 = bt.indicators.SimpleMovingAverage(self.datas[0].close, period=200)
# Order-Tracking
self.order = None
self.buyprice = None
self.buycomm = None
def log(self, txt, dt=None):
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'KAUF AUSGEFÜHRT, Preis: {order.executed.price:.2f}, '
f'Kosten: {order.executed.value:.2f}, Provision: {order.executed.comm:.2f}')
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
else:
self.log(f'VERKauf AUSGEFÜHRT, Preis: {order.executed.price:.2f}, '
f'Kosten: {order.executed.value:.2f}, Provision: {order.executed.comm:.2f}')
self.order = None
def next(self):
# Prüfe ausstehende Orders
if self.order:
return
# Hole aktuelle Preisdaten für KI-Analyse
current_price = self.dataclose[0]
price_data = {
'close': current_price,
'open': self.dataopen[0],
'volume': self.datavolume[0],
'rsi': self.rsi[0],
'change_24h': ((current_price - self.dataclose[-1]) / self.dataclose[-1]) * 100 if len(self.dataclose) > 1 else 0,
'macd': 'bullish' if self.sma50[0] > self.sma200[0] else 'bearish'
}
# KI-Sentiment-Analyse mit HolySheep
if self.params.holy_sheep_client:
result = self.params.holy_sheep_client.analyze_market_sentiment(
symbol=self.params.symbol,
timeframe="1h"
)
# Parse KI-Empfehlung
if result.get('success'):
sentiment_text = result['sentiment'].upper()
ai_signal = 'BUY' if 'BUY' in sentiment_text else ('SELL' if 'SELL' in sentiment_text else 'HOLD')
usage = result.get('usage', {})
cost = (usage.get('prompt_tokens', 0) * 0.42) / 1_000_000 # DeepSeek V3.2: $0.42/MTok
print(f"KI-Signal: {ai_signal}, API-Kosten: ${cost:.6f}")
else:
ai_signal = 'HOLD'
# Strategie-Logik kombinieren
if not self.position:
# Keine Position: Prüfe Kaufsignale
if self.rsi[0] < self.params.rsi_oversold and self.sma50[0] > self.sma200[0]:
self.log(f'KAUF-SIGNAL, RSI: {self.rsi[0]:.2f}, Preis: {current_price:.2f}')
self.order = self.buy()
else:
# Position vorhanden: Prüfe Verkaufssignale
if self.rsi[0] > self.params.rsi_overbought or self.sma50[0] < self.sma200[0]:
self.log(f'VERKauf-SIGNAL, RSI: {self.rsi[0]:.2f}, Preis: {current_price:.2f}')
self.order = self.sell()
def fetch_tardis_data(symbol: str, exchange: str, start_date: datetime, end_date: datetime) -> pd.DataFrame:
"""
Lädt historische Daten von Tardis API.
Tardis bietet minutengenaue OHLCV-Daten für alle major Kryptobörsen.
"""
client = Tardis(api_key="YOUR_TARDIS_API_KEY")
# Tardis API Endpunkt für historische Daten
response = client.get_historical(
exchange=exchange,
symbol=symbol,
start=start_date.isoformat(),
end=end_date.isoformat(),
interval="1m"
)
# Konvertiere zu Pandas DataFrame
df = pd.DataFrame(response)
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)
return df
def run_backtest():
"""
Führt das komplette Backtesting mit HolySheep-KI-Integration durch.
"""
# Initialisiere Cerebro
cerebro = bt.Cerebro(stdstats=True, cheat_on_open=True)
# Lade Tardis-Daten
start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 12, 31)
print("Lade historische Daten von Tardis...")
data = fetch_tardis_data(
symbol='BTCUSD',
exchange='binance',
start_date=start_date,
end_date=end_date
)
# Erstelle Daten-Feed
data_feed = TardisDataFeed(dataname=data)
cerebro.adddata(data_feed)
# Initialisiere HolySheep Client
holy_sheep = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Füge Strategie hinzu
cerebro.addstrategy(
HolySheepStrategy,
holy_sheep_client=holy_sheep,
symbol='BTCUSD',
trade_size=0.1
)
# Broker-Konfiguration
cerebro.broker.setcash(100000) # $100.000 Startkapital
cerebro.broker.setcommission(commission=0.001) # 0.1% Provision
# Position-Sizing
cerebro.addsizer(bt.sizers.FixedSize, stake=0.1)
# Analyzer für Performance-Metriken
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
print(f'Startkapital: ${cerebro.broker.getvalue():.2f}')
# Backtest ausführen
strategies = cerebro.run()
strategy = strategies[0]
print(f'\nEndkapital: ${cerebro.broker.getvalue():.2f}')
print(f'Rendite: {((cerebro.broker.getvalue() - 100000) / 100000) * 100:.2f}%')
# Zeige Charts
cerebro.plot(style='candlestick')
if __name__ == '__main__':
run_backtest()
HolySheep-API: Vollständige Integration mit Error-Handling
import time
import logging
from functools import wraps
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Logging konfigurieren
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def retry_on_failure(max_retries=3, backoff_factor=0.5):
"""
Decorator für automatische Retry-Logik bei API-Fehlern.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
logger.error(f"API-Aufruf nach {max_retries} Versuchen fehlgeschlagen: {e}")
raise
wait_time = backoff_factor * (2 ** attempt)
logger.warning(f"Versuch {attempt + 1} fehlgeschlagen. Warte {wait_time}s...")
time.sleep(wait_time)
return wrapper
return decorator
class HolySheepProductionClient:
"""
Production-ready HolySheep Client mit vollständiger Error-Handling-Logik.
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = self._create_session()
self.request_count = 0
self.total_cost = 0.0
# Preismodell (2026)
self.pricing = {
"gpt-4.1": 8.00, # $8.00/MTok
"claude-sonnet-4.5": 15.00, # $15.00/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok
}
def _create_session(self) -> requests.Session:
"""Erstellt eine Session mit Retry-Mechanismus."""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
def _calculate_cost(self, model: str, usage: dict) -> float:
"""Berechnet die Kosten basierend auf Token-Nutzung."""
price_per_mtok = self.pricing.get(model, 0.42)
total_tokens = usage.get('total_tokens', 0)
cost = (total_tokens / 1_000_000) * price_per_mtok
return cost
@retry_on_failure(max_retries=3)
def chat_completion(self, model: str, messages: list, temperature: float = 0.7, max_tokens: int = 1000) -> dict:
"""
Führt einen Chat-Completion-Aufruf durch.
Args:
model: Modell-ID (deepseek-v3.2, gpt-4.1, etc.)
messages: Liste von Message-Dicts
temperature: Sampling-Temperatur (0-1)
max_tokens: Maximale Token-Anzahl
Returns:
Response-Dict mit KI-Antwort und Nutzungsstatistiken
"""
endpoint = f"{self.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.time()
try:
response = self.session.post(
endpoint,
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
# Latenz messen
latency_ms = (time.time() - start_time) * 1000
# Kosten berechnen
usage = result.get('usage', {})
cost = self._calculate_cost(model, usage)
self.total_cost += cost
self.request_count += 1
return {
"success": True,
"content": result["choices"][0]["message"]["content"],
"model": model,
"latency_ms": round(latency_ms, 2),
"usage": usage,
"cost_usd": round(cost, 6),
"total_requests": self.request_count,
"total_cost_usd": round(self.total_cost, 4)
}
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP-Fehler: {e.response.status_code} - {e.response.text}")
return {
"success": False,
"error": f"HTTP {e.response.status_code}",
"message": e.response.text,
"cost_usd": 0
}
except requests.exceptions.Timeout:
logger.error("Timeout bei API-Anfrage")
return {
"success": False,
"error": "Timeout",
"message": "Anfrage hat das Zeitlimit überschritten",
"cost_usd": 0
}
except Exception as e:
logger.error(f"Unerwarteter Fehler: {str(e)}")
return {
"success": False,
"error": "Unknown",
"message": str(e),
"cost_usd": 0
}
def batch_analysis(self, symbols: list, analysis_type: str = "technical") -> list:
"""
Führt Batch-Analyse für mehrere Symbole durch.
Ideal für Portfolio-Optimierung.
"""
results = []
for symbol in symbols:
logger.info(f"Analysiere {symbol}...")
messages = [
{
"role": "system",
"content": "Du bist ein Krypto-Analyst. Gib präzise, datengestützte Analysen."
},
{
"role": "user",
"content": f"Führe eine {analysis_type}-Analyse für {symbol} durch."
}
]
result = self.chat_completion(
model="deepseek-v3.2", # Kostenoptimal
messages=messages,
temperature=0.3,
max_tokens=500
)
results.append({
"symbol": symbol,
"analysis": result
})
# Rate Limiting: 100ms Pause zwischen Requests
time.sleep(0.1)
return results
def get_cost_summary(self) -> dict:
"""Gibt eine Zusammenfassung der API-Kosten zurück."""
avg_cost_per_request = self.total_cost / self.request_count if self.request_count > 0 else 0
return {
"total_requests": self.request_count,
"total_cost_usd": round(self.total_cost, 4),
"avg_cost_per_request_usd": round(avg_cost_per_request, 6),
"savings_vs_openai": round(self.total_cost * 19, 2) if self.total_cost > 0 else 0, # ~95% Ersparnis
"pricing_model": self.pricing
}
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepProductionClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Einzelne Analyse
result = client.chat_completion(
model="deepseek-v3.2",
messages=[
{"role": "user", "content": "Analysiere den aktuellen Bitcoin-Markttrend für 2024."}
]
)
print(f"Latenz: {result.get('latency_ms')}ms")
print(f"Kosten: ${result.get('cost_usd')}")
print(f"Antwort: {result.get('content')[:200]}...")
# Kostenübersicht
summary = client.get_cost_summary()
print(f"\nGesamtkosten: ${summary['total_cost_usd']}")
print(f" Ersparnis vs. OpenAI: ${summary['savings_vs_openai']}")
Häufige Fehler und Lösungen
Basierend auf meiner jahrelangen Erfahrung mit API-Integrationen in Trading-Systemen habe ich die häufigsten Fallstricke identifiziert und dokumentiert:
1. Authentifizierungsfehler: "401 Unauthorized"
# FEHLER: Falscher API-Key-Format oder abgelaufener Key
Ursache: Key enthält ungültige Zeichen oder ist rotiert
LÖSUNG: Key-Validierung und automatische Rotation
import re
def validate_api_key(key: str) -> bool:
"""Validiert das Format des HolySheep API-Keys."""
if not key or len(key) < 20:
return False
# HolySheep Keys beginnen mit "hs_" gefolgt von Base64
pattern = r'^hs_[A-Za-z0-9_-]{32,}$'
return bool(re.match(pattern, key))
def refresh_api_key(old_key: str) -> str:
"""
Implementiert automatische Key-Rotation.
Bei HolySheep können Sie neue Keys im Dashboard generieren.
"""
if validate_api_key(old_key):
# Key ist noch gültig, keine Aktion nötig
return old_key
# Key ist ungültig - hier würde Ihre Rotation-Logik greifen
raise ValueError("API-Key ist ungültig oder abgelaufen. Bitte generieren Sie einen neuen Key.")
Implementierung in der Client-Klasse
class HolySheepSecureClient(HolySheepAIClient):
def __init__(self, api_key: str):
if not validate_api_key(api_key):
raise ValueError("Ungültiges API-Key-Format")
super().__init__(api_key)
2. Rate-Limiting: "429 Too Many Requests"
# FEHLER: Zu viele Anfragen in kurzer Zeit
Ursache: Keine Implementierung von Rate-Limiting oder Exponential Backoff
LÖSUNG: Token-Bucket-Algorithmus mit automatischer Retry-Logik
import time
import threading
from collections import deque
class TokenBucket:
"""
Token-Bucket-Algorithmus für effektives Rate-Limiting.
HolySheep erlaubt 1000 Requests/Minute im Standard-Tier.
"""
def __init__(self, rate: float, capacity: int):
"""
Args:
rate: Tokens pro Sekunde
capacity: Maximale Token-Kapazität
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self.lock = threading.Lock()
self.request_history = deque(maxlen=1000)
def acquire(self, tokens: int = 1) -> bool:
"""Versucht, Tokens zu akquirieren. Blockiert wenn nötig."""
with self.lock:
now = time.time()
elapsed = now - self.last_update
# Tokens auffüllen
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
self.request_history.append(now)
return True
return False
def wait_and_acquire(self, tokens: int = 1, timeout: float = 60.0) -> bool:
"""Blockiert bis Tokens verfügbar sind oder Timeout erreicht."""
start = time.time()
while time.time() - start < timeout:
if self.acquire(tokens):
return True
# Wartezeit berechnen
wait_time = (tokens - self.tokens) / self.rate
time.sleep(min(wait_time, 1.0))
return False
def get_rate_stats(self) -> dict:
"""Gibt Statistiken über Request-Rate zurück."""
now = time.time()
# Requests der letzten Minute
recent_requests = [t for t in self.request_history if now - t < 60]
return {
"requests_last_minute": len(recent_requests),
"current_tokens": round(self.tokens, 2),
"capacity": self.capacity
}
Verwendung in Production-Client
class RateLimitedHolySheepClient(HolySheepAIClient):
def __init__(self, api_key: str, requests_per_minute: int = 1000):
super().__init__(api_key)
self.bucket = TokenBucket(
rate=requests_per_minute / 60, # Tokens pro Sekunde
capacity=requests_per_minute # Burst-Kapazität
)
def _throttled_request(self, *args, **kwargs):
"""Führt Anfrage mit Rate-Limiting durch."""
if not self.bucket.wait_and_acquire(timeout=30.0):
raise TimeoutError("Rate-Limit Timeout: Bitte warten Sie einen Moment")
return super().chat_completion(*args, **kwargs)
3. Datenqualitätsprobleme mit Tardis
# FEHLER: Fehlende Datenpunkte, Lücken in der Zeitreihe
Ursache: Börsen-Downtimes, API-Limits oder Netzwerkprobleme
LÖSUNG: Datenvalidierung und automatische Lückenschließung
import pandas as pd
import numpy as np
from typing import Tuple
def validate_and_fill_data(df: pd.DataFrame, required_columns: list = None) -> pd.DataFrame:
"""
Validiert Tardis-Daten und füllt fehlende Werte.
Args:
df: DataFrame mit OHLCV-Daten
required_columns: Liste der benötigten Spalten
Returns:
Bereinigter DataFrame mit gefüllten Lücken
"""
if required_columns is None:
required_columns = ['open', 'high', 'low', 'close', 'volume']
# Prüfe auf fehlende Spalten
missing = set(required_columns) - set(df.columns)
if missing:
raise ValueError(f"Fehlende Spalten: {missing}")
# Prüfe Zeitreihen-Kontinuität
df = df.copy()
if 'timestamp' not in df.columns and df.index.name:
df['timestamp'] = df.index
# Resample auf 1-Minute-Basis (Tardis liefert oft unregelmäßige Daten)
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
# Vollständige Zeitreihe erstellen
full_range = pd.date_range(
start=df.index.min(),
end=df.index.max(),
freq='1min'
)
df = df.reindex(full_range)
# Lücken identifizieren
missing_count = df['close'].isna().sum()
if missing_count > 0:
print(f"Warnung: {missing_count} fehlende Datenpunkte gefunden ({missing_count/len(df)*100:.2f}%)")
# Forward-Fill für OHLC (nimmt vorherigen Wert)
df['open'] = df['open'].ffill()
df['high'] = df['high'].ffill().combine(df[['open', 'close']].max(axis=1), max)
df['low'] = df['low'].ffill().combine(df[['open', 'close']].min(axis=1), min)
df['close'] = df['close'].ffill()
df['volume'] = df['volume'].fillna(0)
# Statistische Ausreißer-Erkennung
df = detect_and_handle_outliers(df)
return df
def detect_and_handle_outliers(df: pd.DataFrame, z_threshold: float = 5.0) -> pd.DataFrame:
"""
Erkennt und behandelt statistische Ausreißer in Preisdaten.
Kritisch für zuverlässige Backtesting-Ergebnisse.
"""
df = df.copy()
# Berechne prozentuale Ä