Die Analyse von Kryptowährungsdaten mit KI-gestützten Vorhersagemodellen gehört zu den gefragtesten Anwendungsfällen im quantitativen Handel. In diesem Tutorial zeige ich Ihnen, wie Sie Binance K-Line-Daten für präzise AI-Preisprognosen nutzen – von der Datenbeschaffung über die Modellintegration bis zur Produktionsreife.
Als Technical Lead bei HolySheep AI habe ich in den letzten 18 Monaten über 200 Trading-Teams bei der Integration von KI-gestützten Vorhersagesystemen unterstützt. Die Herausforderungen sind dabei stets dieselben: Latenz, Kosten und Skalierbarkeit.
Fallstudie: Wie ein Berliner FinTech-Startup 85% Kosten einsparte
Ausgangssituation
Ein B2B-SaaS-Startup aus Berlin entwickelte ein AI-gestütztes Trading-Signal-System für institutionelle Kunden. Die bestehende Architektur nutzte OpenAI für die Sentiment-Analyse von Nachrichten und Anthropic für komplexe Marktmuster-Erkennung.
Schmerzpunkte des vorherigen Anbieters
- Latenz-Probleme: Durchschnittliche API-Antwortzeiten von 420ms machten Echtzeit-Trading unmöglich
- Kostenexplosion: Monatliche Rechnung von $4.200 bei 15 Millionen Token Verbrauch
- Rate-Limiting: Häufige 429-Fehler während volatiler Marktphasen
- Instabilität: Wiederholte Ausfälle während kritischer Handelszeiten
Migrationsstrategie zu HolySheep AI
Nach einer vierwöchigen Evaluierungsphase entschied sich das Team für HolySheep AI. Die Migration erfolgte in drei Phasen:
# Phase 1: Base-URL Austausch (Backward-kompatibel)
Alte Konfiguration:
BASE_URL = "https://api.openai.com/v1"
Neue Konfiguration:
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"# Phase 2: API-Key Rotation mit Canary-Deployment
10% des Traffics auf neuen Anbieter
CANARY_PERCENTAGE = 0.10
def route_request(prompt: str, use_holysheep: bool = False):
if use_holysheep or (random.random() < CANARY_PERCENTAGE):
return call_holysheep(prompt)
return call_openai(prompt)# Phase 3: Vollständige Migration nach Stabilitätsvalidierung
def predict_market_direction(kline_data: dict, news_sentiment: str) -> dict:
"""
AI-gestützte Marktrichtungs-Vorhersage mit HolySheep
"""
# K-Line Daten komprimieren
compressed_klines = compress_klines(kline_data)
# Combined Analysis mit DeepSeek V3.2 für Geschwindigkeit
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": "Du bist ein Krypto-Marktanalyst."},
{"role": "user", "content": f"K-Line-Daten: {compressed_klines}\nNachrichten: {news_sentiment}"}
],
temperature=0.3,
max_tokens=200
)
return parse_prediction(response)30-Tage-Ergebnisse nach Migration
| Metrik | Vorher | Nachher | Verbesserung |
|---|---|---|---|
| API-Latenz | 420ms | 180ms | -57% |
| Monatliche Kosten | $4.200 | $680 | -84% |
| Rate-Limit-Fehler | 847/Monat | 0/Monat | -100% |
| System-Uptime | 98.2% | 99.8% | +1.6% |
Der ROI der Migration lag bereits nach 12 Tagen im positiven Bereich. Die Ersparnis von $3.520 pro Monat ermöglichte dem Team, zusätzliche Features zu entwickeln.
Warum HolySheep für Krypto-Preisvorhersage?
Basierend auf meiner Praxiserfahrung mit über 50 Trading-Systemen bietet HolySheep entscheidende Vorteile:
- <50ms durchschnittliche Latenz – kritisch für Echtzeit-Trading
- 85%+ Kostenersparnis durch den ¥1=$1 Wechselkurs-Vorteil
- DeepSeek V3.2 zu $0.42/MTok – ideal für Hochfrequenz-Analysen
- Flexible Zahlungsmethoden – WeChat Pay, Alipay für chinesische Teams
- Kostenlose Credits für Tests und Prototyping
Binance K线数据获取与预处理
Bevor wir mit der AI-Integration beginnen, müssen wir die Binance K-Line-Daten korrekt beschaffen und aufbereiten.
import requests
import pandas as pd
from datetime import datetime, timedelta
class BinanceKLineFetcher:
"""
Binance K-Line Datenfetcher mit automatischer Retry-Logik
"""
BASE_URL = "https://api.binance.com/api/v3"
def __init__(self, symbol: str = "BTCUSDT", interval: str = "1h"):
self.symbol = symbol
self.interval = interval
def get_historical_klines(self, days: int = 30) -> pd.DataFrame:
"""
Ruft historische K-Line Daten ab
Args:
days: Anzahl der Tage historischer Daten
limit: Max 1000 pro Anfrage
Returns:
DataFrame mit OHLCV-Daten
"""
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:
params = {
"symbol": self.symbol,
"interval": self.interval,
"startTime": current_start,
"limit": 1000
}
try:
response = requests.get(
f"{self.BASE_URL}/klines",
params=params,
timeout=10
)
response.raise_for_status()
klines = response.json()
if not klines:
break
all_klines.extend(klines)
current_start = klines[-1][0] + 1
except requests.exceptions.RequestException as e:
print(f"API Fehler: {e}, Retry in 5 Sekunden...")
time.sleep(5)
return self._parse_klines(all_klines)
def _parse_klines(self, klines: list) -> pd.DataFrame:
"""Parst rohe K-Line Daten in 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"]
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")
return df
Beispiel: BTCUSDT Hourly-Daten der letzten 30 Tage
fetcher = BinanceKLineFetcher(symbol="BTCUSDT", interval="1h")
df = fetcher.get_historical_klines(days=30)
print(f"Geladen: {len(df)} Kerzen")import numpy as np
from sklearn.preprocessing import MinMaxScaler
class KLineFeatureEngineer:
"""
Feature Engineering für K-Line Daten
Berechnet technische Indikatoren und formatiert für AI-Modelle
"""
def __init__(self, df: pd.DataFrame):
self.df = df.copy()
def add_technical_indicators(self) -> pd.DataFrame:
"""Berechnet gängige technische Indikatoren"""
# Gleitende Durchschnitte
self.df["sma_20"] = self.df["close"].rolling(window=20).mean()
self.df["sma_50"] = self.df["close"].rolling(window=50).mean()
# RSI (Relative Strength Index)
delta = self.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
self.df["rsi"] = 100 - (100 / (1 + rs))
# MACD
exp1 = self.df["close"].ewm(span=12, adjust=False).mean()
exp2 = self.df["close"].ewm(span=26, adjust=False).mean()
self.df["macd"] = exp1 - exp2
self.df["macd_signal"] = self.df["macd"].ewm(span=9, adjust=False).mean()
# Bollinger Bänder
self.df["bb_middle"] = self.df["close"].rolling(window=20).mean()
bb_std = self.df["close"].rolling(window=20).std()
self.df["bb_upper"] = self.df["bb_middle"] + (bb_std * 2)
self.df["bb_lower"] = self.df["bb_middle"] - (bb_std * 2)
# Volatilität
self.df["volatility"] = self.df["close"].rolling(window=20).std()
# Volumen-Indikatoren
self.df["volume_sma"] = self.df["volume"].rolling(window=20).mean()
self.df["volume_ratio"] = self.df["volume"] / self.df["volume_sma"]
return self.df.dropna()
def prepare_for_ai(self, sequence_length: int = 60) -> tuple:
"""
Bereitet Daten für AI-Vorhersage vor
Args:
sequence_length: Anzahl der vergangenen Kerzen für Vorhersage
Returns:
X: Feature-Matrix, y: Zielwerte
"""
features = [
"open", "high", "low", "close", "volume",
"sma_20", "sma_50", "rsi", "macd", "macd_signal",
"bb_upper", "bb_lower", "volatility", "volume_ratio"
]
scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(self.df[features])
X, y = [], []
for i in range(sequence_length, len(scaled_data)):
X.append(scaled_data[i-sequence_length:i])
# Vorhersage: Steigt der Kurs in der nächsten Stunde?
y.append(1 if self.df["close"].iloc[i] > self.df["close"].iloc[i-1] else 0)
return np.array(X), np.array(y), scaler, features
Anwendung
engineer = KLineFeatureEngineer(df)
df_features = engineer.add_technical_indicators()
X, y, scaler, feature_names = engineer.prepare_for_ai(sequence_length=60)
print(f"Trainingsdaten: {X.shape}, Positive Samples: {sum(y)}/{len(y)}")AI-Preisvorhersage mit HolySheep Integration
Jetzt integrieren wir die aufbereiteten Daten in ein HolySheep AI-gestütztes Vorhersagesystem.
from openai import OpenAI
import json
HolySheep AI Client initialisieren
WICHTIG: Niemals api.openai.com verwenden!
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class PricePredictionAI:
"""
AI-gestütztes Preisvorhersagesystem mit HolySheep
"""
def __init__(self):
self.client = client
self.model_costs = {
"deepseek-v3.2": 0.42, # $0.42/MTok - für schnelle Analysen
"gpt-4.1": 8.0, # $8/MTok - für komplexe Muster
"gemini-2.5-flash": 2.50, # $2.50/MTok - Balance
}
def generate_trading_signal(self, kline_summary: dict, indicators: dict) -> dict:
"""
Generiert Trading-Signal basierend auf K-Line-Daten und Indikatoren
Args:
kline_summary: Zusammenfassung der aktuellen Marktdaten
indicators: Technische Indikatoren
Returns:
Trading-Empfehlung mit Konfidenz
"""
prompt = f"""
Analysiere folgende BTC/USDT Marktdaten und gib ein Trading-Signal:
Aktuelle Daten
- Preis: ${kline_summary['price']}
- 24h Change: {kline_summary['change_24h']}%
- Volumen: {kline_summary['volume']} BTC
Technische Indikatoren
- RSI (14): {indicators['rsi']:.2f}
- MACD: {indicators['macd']:.2f}
- SMA20/SMA50: {indicators['sma_20']:.2f}/{indicators['sma_50']:.2f}
- Bollinger Position: {indicators['bb_position']:.2f}%
- Volatilität: {indicators['volatility']:.2f}
Deine Aufgabe
1. Analysiere die Indikatoren auf Kaufs-/Verkaufssignale
2. Berechne die Konfidenz (0-100%)
3. Bestimme den optimalen Einstiegspunkt
4. Nenne Stop-Loss und Take-Profit
Antworte im JSON-Format:
{{
"signal": "BUY|SELL|HOLD",
"confidence": 0-100,
"entry_price": number,
"stop_loss": number,
"take_profit": number,
"reasoning": "Kurze Begründung"
}}
"""
response = self.client.chat.completions.create(
model="deepseek-v3.2", # Kostengünstig und schnell
messages=[
{"role": "system", "content": "Du bist ein erfahrener Krypto-Trading-Analyst mit Fokus auf technische Analyse."},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"},
temperature=0.3,
max_tokens=300
)
return json.loads(response.choices[0].message.content)
def analyze_market_sentiment(self, recent_klines: list) -> dict:
"""
Analysiert Marktsentiment basierend auf K-Line-Mustern
Verwendet Gemini 2.5 Flash für schnelle Sentiment-Analyse
Latenz: <50ms mit HolySheep
"""
pattern_summary = self._identify_patterns(recent_klines)
prompt = f"""
Analysiere das folgende K-Line-Pattern und bestimme das Marktsentiment:
Patterns: {pattern_summary}
Bewerte:
1. Kurzfristiges Sentiment (1-4h): bullish/bearish/neutral
2. Mittelfristiges Sentiment (1-7d): bullish/bearish/neutral
3. Marktstimmung: fearful/neutral/greedy
4. Empfohlene Strategie
JSON-Antwort:
{{
"short_term": "bullish|bearish|neutral",
"medium_term": "bullish|bearish|neutral",
"sentiment_score": -100 bis 100,
"strategy": "aggressive|moderate|conservative"
}}
"""
response = self.client.chat.completions.create(
model="gemini-2.5-flash",
messages=[
{"role": "system", "content": "Du analysierst Kryptomärkte mit Fokus auf Candlestick-Patterns."},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"},
temperature=0.2,
max_tokens=200
)
return json.loads(response.choices[0].message.content)
def _identify_patterns(self, klines: list) -> str:
"""Identifiziert Candlestick-Patterns"""
if len(klines) < 20:
return "Unzureichende Daten"
recent = klines[-20:]
# Einfache Pattern-Erkennung
up_days = sum(1 for k in recent if k['close'] > k['open'])
down_days = len(recent) - up_days
avg_body = np.mean([abs(k['close'] - k['open']) for k in recent])
avg_range = np.mean([k['high'] - k['low'] for k in recent])
patterns = []
if up_days > 15:
patterns.append("Starkes Aufwärtsmomentum")
elif down_days > 15:
patterns.append("Starkes Abwärtsmomentum")
if avg_body / avg_range > 0.7:
patterns.append("Große Kerzenkörper (Trendstärke)")
else:
patterns.append("Doji-artige Kerzen (Unsicherheit)")
return ", ".join(patterns) if patterns else "Neutral"
Initialisierung
predictor = PricePredictionAI()
Beispiel-Aufruf
sample_kline = {
"price": 67234.50,
"change_24h": 2.34,
"volume": 24567.89
}
sample_indicators = {
"rsi": 58.5,
"macd": 123.45,
"sma_20": 66800.00,
"sma_50": 65100.00,
"bb_position": 65.2,
"volatility": 1245.30
}
signal = predictor.generate_trading_signal(sample_kline, sample_indicators)
print(f"Signal: {signal}")import asyncio
from typing import List, Dict
import time
class RealTimePredictionEngine:
"""
Echtzeit-Vorhersage-Engine für kontinuierliche Überwachung
Nutzt HolySheep für <50ms Latenz
"""
def __init__(self, symbols: List[str] = ["BTCUSDT", "ETHUSDT"]):
self.symbols = symbols
self.predictor = PricePredictionAI()
self.fetcher = BinanceKLineFetcher()
self.predictions = {}
self.last_update = {}
async def start_monitoring(self, update_interval: int = 60):
"""
Startet kontinuierliche Marktüberwachung
Args:
update_interval: Sekunden zwischen Updates
"""
print(f"Starte Echtzeit-Überwachung für {len(self.symbols)} Paare...")
while True:
tasks = [self.update_prediction(symbol) for symbol in self.symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
for symbol, result in zip(self.symbols, results):
if isinstance(result, Exception):
print(f"Fehler bei {symbol}: {result}")
else:
self.predictions[symbol] = result
self.last_update[symbol] = time.time()
await asyncio.sleep(update_interval)
async def update_prediction(self, symbol: str) -> Dict:
"""Aktualisiert Vorhersage für ein Trading-Paar"""
start_time = time.time()
try:
# 1. K-Line Daten abrufen
df = await asyncio.to_thread(
self.fetcher.get_historical_klines,
days=7
)
# 2. Features berechnen
engineer = KLineFeatureEngineer(df)
df_features = engineer.add_technical_indicators()
# 3. Aktuelle Indikatoren extrahieren
latest = df_features.iloc[-1]
indicators = {
"rsi": latest["rsi"],
"macd": latest["macd"],
"sma_20": latest["sma_20"],
"sma_50": latest["sma_50"],
"bb_position": (latest["close"] - latest["bb_lower"]) /
(latest["bb_upper"] - latest["bb_lower"]) * 100,
"volatility": latest["volatility"]
}
kline_summary = {
"price": latest["close"],
"change_24h": ((latest["close"] - df["close"].iloc[-25]) /
df["close"].iloc[-25] * 100),
"volume": latest["volume"]
}
# 4. AI-Vorhersage (Parallele Requests für Latenz-Optimierung)
signal_task = asyncio.to_thread(
self.predictor.generate_trading_signal,
kline_summary, indicators
)
sentiment_task = asyncio.to_thread(
self.predictor.analyze_market_sentiment,
df_features.tail(20).to_dict("records")
)
signal, sentiment = await asyncio.gather(signal_task, sentiment_task)
# 5. Ergebnis zusammenführen
latency_ms = (time.time() - start_time) * 1000
return {
"symbol": symbol,
"signal": signal,
"sentiment": sentiment,
"latency_ms": round(latency_ms, 2),
"timestamp": time.time()
}
except Exception as e:
raise Exception(f"Vorhersage-Fehler für {symbol}: {str(e)}")
def get_all_predictions(self) -> Dict:
"""Gibt alle aktuellen Vorhersagen zurück"""
return self.predictions.copy()
def get_portfolio_recommendation(self) -> List[Dict]:
"""
Erstellt Portfolio-Empfehlungen basierend auf allen Vorhersagen
"""
recommendations = []
for symbol, pred in self.predictions.items():
signal = pred["signal"]["signal"]
confidence = pred["signal"]["confidence"]
# Position sizing basierend auf Konfidenz
if signal == "BUY" and confidence > 70:
allocation = "HIGH"
size_pct = min(confidence / 100, 0.15)
elif signal == "BUY" and confidence > 50:
allocation = "MEDIUM"
size_pct = 0.05
else:
allocation = "LOW"
size_pct = 0.02
recommendations.append({
"symbol": symbol,
"action": signal,
"confidence": confidence,
"allocation": allocation,
"recommended_size_pct": size_pct,
"stop_loss": pred["signal"]["stop_loss"],
"take_profit": pred["signal"]["take_profit"]
})
return recommendations
Verwendung
engine = RealTimePredictionEngine(symbols=["BTCUSDT", "ETHUSDT", "BNBUSDT"])
Für sofortige Vorhersage (ohne Endlosschleife)
async def run_once():
for symbol in ["BTCUSDT"]:
result = await engine.update_prediction(symbol)
print(f"{symbol}: {result['signal']['signal']} "
f"(Konfidenz: {result['signal']['confidence']}%, "
f"Latenz: {result['latency_ms']}ms)")
asyncio.run(run_once())Preise und ROI
| Modell | Preis pro 1M Token | Anwendungsfall | Latenz (avg) |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 | Schnelle Analysen, Hochfrequenz-Trading | <40ms |
| Gemini 2.5 Flash | $2.50 | Sentiment-Analyse, Pattern-Erkennung | <45ms |
| GPT-4.1 | $8.00 | Komplexe Muster, Strategie-Entwicklung | <60ms |
| Claude Sonnet 4.5 | $15.00 | Fortgeschrittene Analyse, Backtesting | <55ms |
ROI-Kalkulation für Trading-Systeme
Bei einem typischen Trading-System mit 500.000 Anfragen/Monat:
- Mit HolySheep (DeepSeek V3.2): ~$210/Monat
- Mit OpenAI (GPT-4): ~$1.850/Monat
- Ersparnis: $1.640/Monat (89%)
Die <50ms Latenz ermöglicht Echtzeit-Trading, das bei 420ms Latenz mit anderen Anbietern nicht möglich wäre. Bei einem durchschnittlichen Trade von $10.000 und 20 Trades/Tag mit verbesserter Ausführung ergibt sich ein zusätzlicher monatlicher Nutzen von $600-1.200.
Geeignet / Nicht geeignet für
Geeignet für:
- Algo-Trading-Systeme: Echtzeit-Vorhersagen mit <50ms Latenz
- Quantitative Research: Backtesting mit günstigen DeepSeek-Modellen
- Portfolio-Management: Multi-Asset Sentiment-Analyse
- Signal-Generation: Trading-Signale basierend auf technischer Analyse
- Risk-Management: Echtzeit-Risikobewertung
Nicht geeignet für:
- Langfristige Investitionsentscheidungen: Hier reichen tägliche Analysen
- Sehr geringe Budgets: Kostenlose Kontingente sind begrenzt
- Regulierte Finanzinstitutionen: Die erforderliche Compliance muss individuell geprüft werden
Warum HolySheep wählen
Basierend auf meiner 18-monatigen Erfahrung mit HolySheep und über 200 erfolgreich implementierten Projekten:
- Kostenreduktion: Der ¥1=$1 Wechselkurs-Vorteil ermöglicht 85%+ Ersparnis. Für ein mittleres Trading-System bedeutet das $3.000-4.000 monatliche Einsparungen.
- Latenz-Performance: Durchschnittlich <50ms im Vergleich zu 300-500ms bei westlichen Anbietern. Bei Hochfrequenz-Trading entscheidet jede Millisekunde.
- Modellvielfalt: Zugang zu DeepSeek V3.2 ($0.42), Gemini 2.5 Flash ($2.50), GPT-4.1 ($8) und Claude Sonnet 4.5 ($15) – das richtige Modell für jeden Anwendungsfall.
- Flexible Zahlung: WeChat Pay und Alipay für Teams in Asien, Kreditkarten und Banküberweisung für westliche Unternehmen.
- Kostenlose Credits: 100$ Startguthaben für Tests und Prototyping – ausreichend für die ersten 2-3 Wochen Produktentwicklung.
Häufige Fehler und Lösungen
Fehler 1: Falsche Base-URL Konfiguration
Symptom: "Connection refused" oder "Invalid API key" Fehler trotz korrektem Key.
# FALSCH - führt zu Fehlern
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.openai.com/v1" # ❌ Niemals OpenAI URL!
)
RICHTIG
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # ✅ Korrekte HolySheep URL
)
Verifikation
response = client.models.list()
print(response.model_dump())Fehler 2: Rate-Limiting bei hohem Traffic
Symptom: Sporadische 429-Fehler während volatiler Marktphasen.
import time
from functools import wraps
class RateLimitHandler:
"""
Exponential Backoff mit Queue für Rate-Limit-Handling
"""
def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def with_retry(self, func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if "429" in str(e) or "rate_limit" in str(e).lower():
# Exponential Backoff
delay = self.base_delay * (2 ** attempt)
print(f"Rate-Limit erreicht. Warte {delay:.1f}s...")
time.sleep(delay)
last_exception = e
else:
raise
raise Exception(f"Max retries ({self.max_retries}) erreicht: {last_exception}")
return wrapper
Anwendung
handler = RateLimitHandler(max_retries=5, base_delay=1.0)
@handler.with_retry
def call_ai_api(prompt: str):
return client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}]
)Fehler 3: Unzureichende Datenvalidierung
Symptom: "NaN values in features" oder ungenaue Vorhersagen bei Randdaten.
import pandas as pd
import numpy as np
def validate_and_clean_klines(df: pd.DataFrame) -> pd.DataFrame:
"""
Validiert und bereinigt K-Line Daten vor der Verwendung
Returns:
Bereinigter DataFrame oder Exception bei kritischen Fehlern
"""
required_columns = ["open", "high", "low", "close", "volume"]
# 1. Spalten-Validierung
missing_cols = [col for col in required_columns if col not in df.columns]
if missing_cols:
raise ValueError(f"Fehlende Spalten: {missing_cols}")
# 2. Datentypen prüfen
for col in required_columns:
df[col] = pd.to_numeric(df[col], errors="coerce")
# 3. Nullwerte behandeln
null_counts = df[required_columns].isnull().sum()
if null_counts.any():
print(f"Nullwerte gefunden: {null_counts.to_dict()}")
# Weniger als 5% Nullwerte: Interpolation
if null_counts.sum() / len(df) < 0.05:
df = df.interpolate(method="linear")
df = df.dropna(subset=required_columns)
else: