Stellen Sie sich vor: Sie sind ein Indie-Entwickler, der seit drei Monaten an einer Bitcoin-Trading-Strategie arbeitet. Ihre Hypothese basiert auf volumengewichteten Durchschnittspreisen und Momentum-Indikatoren. Bevor Sie echtes Kapital riskieren, möchten Sie die Strategie gegen zwei Jahre historischer Daten validieren. Das Problem? Rohdaten von Börsen sind unstrukturiert, die Verarbeitung kostspielig, und herkömmliche Backtesting-Frameworks liefern keine Einblicke in die semantische Bedeutung von Marktmustern.

In diesem Tutorial zeige ich Ihnen, wie Sie Tardis.dev Historical Ticker Data mit KI-gestützter Analyse kombinieren, um robuste Strategie-Backtests durchzuführen. Sie lernen, wie Sie historische Marktdaten abrufen, mit Large Language Models analysieren und so zu fundierten Trading-Entscheidungen gelangen.

Was ist Tardis.dev und warum eignet es sich für Backtesting?

Tardis.dev ist ein Dienstleister für historische Marktdaten im Kryptobereich. Im Gegensatz zu vielen Konkurrenten bietet Tardis.dev:

Für unser Backtesting-Projekt nutzen wir speziell die Ticker-Daten, die individuelle Handelsaufträge mit Preis, Volumen und Timestamp erfassen – ideal für volumengewichtete Strategien.

Voraussetzungen und Setup

Bevor wir beginnen, benötigen Sie:

# Installation der erforderlichen Pakete
pip install requests pandas numpy python-dateutil


Für fortgeschrittene Visualisierung (optional)

pip install matplotlib plotly


Überprüfung der Installation

python -c "import requests, pandas; print('Setup erfolgreich!')"

Schritt 1: Historische Daten von Tardis.dev abrufen

Der erste Schritt besteht darin, relevante historische Ticker-Daten zu beschaffen. Tardis.dev bietet verschiedene Endpunkte für unterschiedliche Datentypen. Für unser Backtesting konzentrieren wir uns auf aggregierte Ticker-Daten, die Preisbewegungen über Zeitintervalle zusammenfassen.

import requests
import pandas as pd
from datetime import datetime, timedelta
import time

class TardisClient:
    """
    Client für Tardis.dev Historical Data API
    Ermöglicht das Abrufen von Ticker-Daten für Backtesting
    """
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_token: str):
        self.api_token = api_token
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_token}'
        })
    
    def get_symbols(self, exchange: str) -> list:
        """
        Listet alle verfügbaren Symbole einer Börse auf
        """
        url = f"{self.BASE_URL}/exchanges/{exchange}/symbols"
        response = self.session.get(url)
        response.raise_for_status()
        return response.json()
    
    def get_ticker_data(self, exchange: str, symbol: str, 
                        start_date: str, end_date: str,
                        limit: int = 10000) -> pd.DataFrame:
        """
        Ruft historische Ticker-Daten für ein Symbol ab
        
        Args:
            exchange: Börsenname (z.B. 'binance', 'coinbase')
            symbol: Trading-Paar (z.B. 'BTC-USDT')
            start_date: Startdatum im Format 'YYYY-MM-DD'
            end_date: Enddatum im Format 'YYYY-MM-DD'
            limit: Maximale Anzahl an Datensätzen pro Anfrage
        
        Returns:
            DataFrame mit Ticker-Daten
        """
        url = f"{self.BASE_URL}/feeds/{exchange}:{symbol}"
        params = {
            'from': start_date,
            'to': end_date,
            'limit': limit
        }
        
        all_tickers = []
        last_id = None
        
        while True:
            if last_id:
                params['after'] = last_id
            
            response = self.session.get(url, params=params)
            response.raise_for_status()
            
            data = response.json()
            if not data or 'tickers' not in data:
                break
                
            all_tickers.extend(data['tickers'])
            
            # Pagination: Tardis.dev limitiert Antworten
            if len(data['tickers']) < limit:
                break
                
            last_id = data['tickers'][-1].get('id')
            time.sleep(0.1)  # Rate Limiting respektieren
            
            print(f"Abgerufen: {len(all_tickers)} Ticker...")
        
        # DataFrame erstellen
        df = pd.DataFrame(all_tickers)
        
        if not df.empty:
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df = df.sort_values('timestamp')
        
        return df


Beispiel-Nutzung

TARDIS_API_KEY = "ihr_tardis_api_token"  # Ersetzen Sie mit Ihrem Token
client = TardisClient(TARDIS_API_KEY)


BTC/USDT Daten von Binance für Januar 2024 abrufen

btc_data = client.get_ticker_data(
    exchange='binance',
    symbol='BTC-USDT',
    start_date='2024-01-01',
    end_date='2024-01-31'
)

print(f"\nAbgerufene Datensätze: {len(btc_data)}")
print(btc_data.head())

Schritt 2: Strategie-Signale aus Marktdaten extrahieren

Mit den rohen Ticker-Daten können wir nun technische Indikatoren berechnen und Trading-Signale generieren. Für unser Beispiel implementieren wir eine Mean-Reversion-Strategie mit Volumenbestätigung.

import numpy as np

class TradingStrategy:
    """
    Mean-Reversion Strategie mit Volumenanalyse
    Generiert Kauf-/Verkaufssignale basierend auf:
    - Bollinger Bands für Preisanomalien
    - Volumenveränderungen zur Bestätigung
    """
    
    def __init__(self, window: int = 20, std_dev: float = 2.0,
                 volume_threshold: float = 1.5):
        self.window = window
        self.std_dev = std_dev
        self.volume_threshold = volume_threshold
    
    def calculate_bollinger_bands(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Berechnet Bollinger Bands
        """
        df = df.copy()
        df['SMA'] = df['price'].rolling(window=self.window).mean()
        df['STD'] = df['price'].rolling(window=self.window).std()
        df['Upper_Band'] = df['SMA'] + (df['STD'] * self.std_dev)
        df['Lower_Band'] = df['SMA'] - (df['STD'] * self.std_dev)
        return df
    
    def calculate_volume_ma(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Berechnet gleitenden Volumendurchschnitt
        """
        df = df.copy()
        df['Volume_MA'] = df['volume'].rolling(window=self.window).mean()
        df['Volume_Ratio'] = df['volume'] / df['Volume_MA']
        return df
    
    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Generiert Trading-Signale
        BUY: Preis unter unterem Bollinger Band + hohes Volumen
        SELL: Preis über oberem Bollinger Band + hohes Volumen
        """
        df = self.calculate_bollinger_bands(df)
        df = self.calculate_volume_ma(df)
        
        # Signale definieren
        df['Signal'] = 'HOLD'
        
        # Kaufsignal: Preis unter unterem Band UND Volumen bestätigt
        buy_condition = (
            (df['price'] < df['Lower_Band']) & 
            (df['Volume_Ratio'] > self.volume_threshold)
        )
        df.loc[buy_condition, 'Signal'] = 'BUY'
        
        # Verkaufssignal: Preis über oberem Band UND Volumen bestätigt
        sell_condition = (
            (df['price'] > df['Upper_Band']) & 
            (df['Volume_Ratio'] > self.volume_threshold)
        )
        df.loc[sell_condition, 'Signal'] = 'SELL'
        
        return df
    
    def backtest(self, df: pd.DataFrame, initial_capital: float = 10000) -> dict:
        """
        Führt Backtest der Strategie durch
        
        Returns:
            Dictionary mit Performance-Metriken
        """
        df = self.generate_signals(df)
        
        capital = initial_capital
        position = 0  # Anzahl BTC
        trades = []
        
        for idx, row in df.iterrows():
            if row['Signal'] == 'BUY' and position == 0:
                # All-in kaufen
                position = capital / row['price']
                trades.append({
                    'type': 'BUY',
                    'price': row['price'],
                    'timestamp': row['timestamp'],
                    'capital': capital
                })
                capital = 0
                
            elif row['Signal'] == 'SELL' and position > 0:
                # Alles verkaufen
                capital = position * row['price']
                trades.append({
                    'type': 'SELL',
                    'price': row['price'],
                    'timestamp': row['timestamp'],
                    'capital': capital
                })
                position = 0
        
        # Finale Bewertung
        final_value = capital + (position * df.iloc[-1]['price'])
        total_return = ((final_value - initial_capital) / initial_capital) * 100
        
        # Buy & Hold Vergleich
        first_price = df.iloc[0]['price']
        last_price = df.iloc[-1]['price']
        buy_hold_return = ((last_price - first_price) / first_price) * 100
        
        return {
            'initial_capital': initial_capital,
            'final_value': final_value,
            'total_return_pct': total_return,
            'buy_hold_return_pct': buy_hold_return,
            'num_trades': len(trades),
            'win_rate': self._calculate_win_rate(trades),
            'trades': trades
        }
    
    def _calculate_win_rate(self, trades: list) -> float:
        """
        Berechnet Trefferquote aus Trades
        """
        if len(trades) < 2:
            return 0.0
        
        wins = 0
        buy_price = None
        
        for trade in trades:
            if trade['type'] == 'BUY':
                buy_price = trade['price']
            elif trade['type'] == 'SELL' and buy_price:
                if trade['price'] > buy_price:
                    wins += 1
                buy_price = None
        
        sell_trades = sum(1 for t in trades if t['type'] == 'SELL')
        return (wins / sell_trades * 100) if sell_trades > 0 else 0


Backtest ausführen

strategy = TradingStrategy(window=20, std_dev=2.0, volume_threshold=1.5)
results = strategy.backtest(btc_data, initial_capital=10000)

print("=" * 50)
print("BACKTEST ERGEBNISSE")
print("=" * 50)
print(f"Startkapital: ${results['initial_capital']:,.2f}")
print(f"Endwert: ${results['final_value']:,.2f}")
print(f"Rendite Strategie: {results['total_return_pct']:.2f}%")
print(f"Rendite Buy&Hold: {results['buy_hold_return_pct']:.2f}")
print(f"Anzahl Trades: {results['num_trades']}")
print(f"Win Rate: {results['win_rate']:.1f}%")

Schritt 3: KI-gestützte Marktanalyse mit HolySheep AI

Der Backtest liefert quantitative Ergebnisse. Aber was bedeuten diese Muster semantisch? Hier kommt HolySheep AI ins Spiel. Mit DeepSeek V3.2 für nur $0.42 pro Million Tokens (85%+ günstiger als OpenAI) können wir die Marktdaten intelligent analysieren lassen.

Integration von HolySheep AI für Sentiment-Analyse

import requests
import json

class HolySheepAnalyzer:
    """
    KI-gestützte Marktanalyse mit HolySheep AI API
    Nutzt DeepSeek V3.2 für kostengünstige und schnelle Analyse
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.model = "deepseek-v3.2"
    
    def analyze_market_pattern(self, df: pd.DataFrame, 
                              strategy_results: dict) -> dict:
        """
        Analysiert Marktdaten und Backtest-Ergebnisse mit KI
        
        Args:
            df: DataFrame mit Marktdaten
            strategy_results: Dictionary mit Backtest-Ergebnissen
        
        Returns:
            KI-gestützte Analyse als Dictionary
        """
        # Zusammenfassung der Daten erstellen
        summary = self._create_data_summary(df, strategy_results)
        
        prompt = f"""Analysiere die folgenden Krypto-Marktdaten und Backtest-Ergebnisse:

DATENZUSAMMENFASSUNG:
{summary}

AUFGABE:
1. Identifiziere die wichtigsten Muster in den Daten
2. Erkläre mögliche Gründe für Gewinne/Verluste der Strategie
3. Gibt Verbesserungsvorschläge für die Strategie?
4. Welche externen Faktoren könnten die Ergebnisse beeinflusst haben?

Antworte auf Deutsch in strukturiertem Format."""
        
        response = self._call_llm(prompt)
        return response
    
    def generate_trading_insights(self, df: pd.DataFrame) -> dict:
        """
        Generiert Trading-Einblicke basierend auf historischen Daten
        """
        # Die letzten 100 Datenpunkte für Analyse
        recent_data = df.tail(100)
        
        prompt = f"""Analysiere die folgenden recenten Bitcoin-Kursdaten:

Letzte Kurse:
{recent_data[['timestamp', 'price', 'volume']].to_string()}

Volatilität der letzten 20 Perioden: {df['price'].tail(20).std():.2f}
Durchschnittsvolumen: {df['volume'].tail(20).mean():.2f}

Erkläre:
1. Kurzfristige Preistrends
2. Volumenmuster und deren Bedeutung
3. Mögliche Unterstützungs-/Widerstandsniveaus
4. Risikoeinschätzung für kurzfristige Positionen

Antworte auf Deutsch mit klaren Handlungsempfehlungen."""
        
        return self._call_llm(prompt)
    
    def _create_data_summary(self, df: pd.DataFrame, 
                            strategy_results: dict) -> str:
        """Erstellt eine文本liche Zusammenfassung der Daten"""
        return f"""
Zeitraum: {df['timestamp'].min()} bis {df['timestamp'].max()}
Anzahl Datenpunkte: {len(df)}
Preisrange: ${df['price'].min():,.2f} - ${df['price'].max():,.2f}
Volatilität (StdAbw): ${df['price'].std():,.2f}
Durchschnittsvolumen: {df['volume'].mean():,.2f}

BACKTEST ERGEBNISSE:
- Strategie-Rendite: {strategy_results['total_return_pct']:.2f}%
- Buy&Hold-Rendite: {strategy_results['buy_hold_return_pct']:.2f}%
- Anzahl Trades: {strategy_results['num_trades']}
- Win Rate: {strategy_results['win_rate']:.1f}%
"""
    
    def _call_llm(self, prompt: str, max_tokens: int = 1000) -> dict:
        """
        Ruft die HolySheep AI API auf
        """
        url = f"{self.base_url}/chat/completions"
        
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'model': self.model,
            'messages': [
                {
                    'role': 'system',
                    'content': 'Du bist ein erfahrener Krypto-Trading-Analyst mit Fokus auf quantitative Analyse und Sentiment-Erkennung.'
                },
                {
                    'role': 'user',
                    'content': prompt
                }
            ],
            'max_tokens': max_tokens,
            'temperature': 0.7
        }
        
        response = requests.post(url, headers=headers, json=payload)
        
        if response.status_code != 200:
            raise Exception(f"API Fehler: {response.status_code} - {response.text}")
        
        result = response.json()
        
        return {
            'content': result['choices'][0]['message']['content'],
            'usage': result.get('usage', {}),
            'model': result.get('model', self.model)
        }


HolySheep AI Analyzer initialisieren

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
analyzer = HolySheepAnalyzer(HOLYSHEEP_API_KEY)


Marktanalyse durchführen

print("Führe KI-gestützte Analyse durch...")
print("Modell: DeepSeek V3.2 ($0.42/MTok bei HolySheep)")
print("-" * 50)

market_analysis = analyzer.analyze_market_pattern(btc_data, results)
print("\n📊 KI-ANALYSE ERGEBNIS:")
print(market_analysis['content'])


Token-Verbrauch anzeigen

if 'usage' in market_analysis:
    usage = market_analysis['usage']
    input_tokens = usage.get('prompt_tokens', 0)
    output_tokens = usage.get('completion_tokens', 0)
    total_tokens = usage.get('total_tokens', 0)
    
    # Kosten berechnen (DeepSeek V3.2: $0.42/MTok Input, $0.42/MTok Output)
    cost_per_mtok = 0.42
    estimated_cost = (total_tokens / 1_000_000) * cost_per_mtok
    
    print("\n💰 TOKEN-VERBRAUCH:")
    print(f"Input Tokens: {input_tokens:,}")
    print(f"Output Tokens: {output_tokens:,}")
    print(f"Gesamt Tokens: {total_tokens:,}")
    print(f"Geschätzte Kosten: ${estimated_cost:.4f}")

Schritt 4: Vollständige Backtesting-Pipeline

Lassen Sie uns alle Komponenten in einer vollständigen Pipeline zusammenführen, die Sie direkt in Ihrem Projekt verwenden können.

 Dict:
        """
        Führt vollständigen Backtest mit KI-Analyse durch
        """
        print(f"\n🚀 STARTE BACKTEST")
        print(f"Börse: {exchange}, Symbol: {symbol}")
        print(f"Zeitraum: {start_date} bis {end_date}")
        print("-" * 40)
        
        # 1. Daten abrufen
        print("📥 Lade historische Daten von Tardis.dev...")
        df = self.tardis.get_ticker_data(exchange, symbol, start_date, end_date)
        print(f"   ✓ {len(df)} Datenpunkte geladen")
        
        # 2. Strategie initialisieren
        print("📈 Initialisiere Trading-Strategie...")
        strategy = TradingStrategy(**strategy_params)
        
        # 3. Backtest durchführen
        print("🔄 Führe Backtest durch...")
        results = strategy.backtest(df)
        print(f"   ✓ Backtest abgeschlossen")
        
        # 4. KI-Analyse (nur wenn HolySheep Key vorhanden)
        print("🤖 Starte KI-gestützte Analyse mit HolySheep AI...")
        print("   Modell: DeepSeek V3.2 ($0.42/MTok)")
        
        try:
            analysis = self.holysheep.analyze_market_pattern(df, results)
            results['ki_analyse'] = analysis['content']
            results['token_usage'] = analysis.get('usage', {})
        except Exception as e:
            print(f"   ⚠ KI-Analyse fehlgeschlagen: {e}")
            results['ki_analyse'] = None
        
        # 5. Zusammenfassung
        self._print_summary(results)
        
        return results
    
    def _print_summary(self, results: Dict):
        """Gibt eine formatierte Zusammenfassung aus"""
        print("\n" + "=" * 50)
        print("📊 BACKTEST ZUSAMMENFASSUNG")
        print("=" * 50)
        print(f"Startkapital:      ${results['initial_capital']:,.2f}")
        print(f"Endwert:           ${results['final_value']:,.2f}")
        print(f"Rendite:           {results['total_return_pct']:+.2f}%")
        print(f"Buy & Hold:        {results['buy_hold_return_pct']:+.2f}%")
        print(f"Überrendite:       {results['total_return_pct'] - results['buy_hold_return_pct']:+.2f}%")
        print(f"Trades:            {results['num_trades']}")
        print(f"Win Rate:          {results['win_rate']:.1f}%")
        
        if results.get('token_usage'):
            tokens = results['token_usage'].get('total_tokens', 0)
            cost = (tokens / 1_000_000) * 0.42
            print(f"KI-Kosten:         ${cost:.4f}")
        
        print("=" * 50)


============== HAUPTPROGRAMM ==============

if __name__ == "__main__":
    # API Keys (durch Ihre eigenen ersetzen)
    HOLYSHEEP_KEY = "YOUR_HOLYSHEEP_API_KEY"
    TARDIS_KEY = "ihr_tardis_token"
    
    # Initialize Backtester
    backtester = CryptoBacktester(HOLYSHEEP_KEY, TARDIS_KEY)
    
    # Parameter für Strategie
    strategy_params = {
        'window': 20,
        'std_dev': 2.0,
        'volume_threshold': 1.5
    }
    
    # Backtest ausführen
    results = backtester.run_full_backtest(
        exchange='binance',
        symbol='BTC-USDT',
        start_date='2024-01-01',
        end_date='2024-03-31',
        strategy_params=strategy_params
    )
    
    # KI-Analyse anzeigen
    if results.get('ki_analyse'):
        print("\n🤖 KI-ANALYSE:")
        print("-" * 50)
        print(results['ki_analyse'])

Häufige Fehler und Lösungen

1. Tardis.dev API Rate Limiting

Problem: Bei zu vielen Anfragen erhalten Sie einen 429 Too Many Requests Fehler.

# Lösung: Implementieren Sie Exponential Backoff
import time
import requests

def fetch_with_retry(url: str, max_retries: int = 5, 
                     base_delay: float = 1.0) -> requests.Response:
    """
    Ruft Daten mit automatischem Retry und Exponential Backoff ab
    """
    for attempt in range(max_retries):
        try:
            response = requests.get(url)
            
            if response.status_code == 200:
                return response
            elif response.status_code == 429:
                # Rate Limited - warten mit Exponential Backoff
                wait_time = base_delay * (2 ** attempt)
                print(f"Rate Limited. Warte {wait_time:.1f}s...")
                time.sleep(wait_time)
            else:
                response.raise_for_status()
                
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            wait_time = base_delay * (2 ** attempt)
            print(f"Fehler: {e}. Retry in {wait_time:.1f}s...")
            time.sleep(wait_time)
    
    raise Exception(f"Max retries ({max_retries}) erreicht")

2. HeilSheep API Authentifizierungsfehler

Problem: 401 Unauthorized bei HolySheep API-Aufrufen.

# Lösung: Prüfen Sie den API-Key und Header-Format
import os

def validate_holysheep_key(api_key: str) -> bool:
    """
    Validiert den HolySheep API Key vor der Nutzung
    """
    if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
        print("❌ Fehler: Bitte setzen Sie Ihren HolySheep API Key")
        print("   Erstellen Sie einen Key unter: https://www.holysheep.ai/register")
        return False
    
    if len(api_key) < 20:
        print("❌ Fehler: API Key scheint zu kurz zu sein")
        return False
    
    return True


Verwendung vor API-Aufruf

HOLYSHEEP_KEY = os.environ.get('HOLYSHEEP_API_KEY', 'YOUR_HOLYSHEEP_API_KEY')

if validate_holysheep_key(HOLYSHEEP_KEY):
    analyzer = HolySheepAnalyzer(HOLYSHEEP_KEY)
    # ... weiter mit Analyse
else:
    print("Fallback: Nur quantitativer Backtest ohne KI-Analyse")

3. Datenqualitätsprobleme bei Backtesting

Problem: Fehlende Datenpunkte oder Outliers verfälschen Backtest-Ergebnisse.

import pandas as pd
import numpy as np

def clean_backtest_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Bereinigt Daten für zuverlässige Backtests
    
    Behandelt:
    - Fehlende Werte
    - Extreme Outliers
    - Unregelmäßige Zeitabstände
    """
    df = df.copy()
    original_len = len(df)
    
    # 1. Fehlende Werte behandeln
    df = df.dropna(subset=['price', 'volume', 'timestamp'])
    
    # 2. Null- und Negativpreise entfernen
    df = df[df['price'] > 0]
    df = df[df['volume'] >= 0]
    
    # 3. Extreme Outliers mit IQR-Methode entfernen
    Q1 = df['price'].quantile(0.25)
    Q3 = df['price'].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 3 * IQR  # 3-facher IQR für extreme Outliers
    upper_bound = Q3 + 3 * IQR
    
    df = df[(df['price'] >= lower_bound) & (df['price'] <= upper_bound)]
    
    # 4. Zeitliche Lücken identifizieren (> 1 Stunde)
    df = df.sort_values('timestamp')
    df['time_diff'] = df['timestamp'].diff()
    large_gaps = df[df['time_diff'] > pd.Timedelta(hours=1)]
    
    if len(large_gaps) > 0:
        print(f"⚠️ Warnung: {len(large_gaps)} große Zeitlücken gefunden")
        print(f"   Erste Lücke: {large_gaps.iloc[0]['timestamp']}")
    
    # 5. Nach Zeit sortieren und Index zurücksetzen
    df = df.sort_values('timestamp').reset_index(drop=True)
    
    removed = original_len - len(df)
    if removed > 0:
        print(f"✓ Bereinigt: {removed} Datensätze entfernt ({removed/original_len*100:.1f}%)")
    
    return df


Verwendung im Backtest

cleaned_data = clean_backtest_data(btc_data)
print(f"Bereinigte Daten: {len(cleaned_data)} Punkte")

4. Falsche Periodenberechnung bei Strategien

Problem: Look-ahead Bias oder unberücksichtigte Transaktionskosten.

from dataclasses import dataclass

@dataclass
class BacktestConfig:
    """Konfiguration für realistischen Backtest"""
    initial_capital: float = 10000.0
    transaction_fee_pct: float = 0.1  # Binance: 0.1% Maker Fee
    slippage_pct: float = 0.05  # 0.05% Slippage
    min_trade_value: float = 10.0  # Mindestorderwert

class RealisticBacktester(TradingStrategy):
    """
    Erweiterter Backtester mit realistischen Kosten
    Beugt Look-ahead Bias vor
    """
    
    def __init__(self, config: BacktestConfig, **strategy_params):
        super().__init__(**strategy_params)
        self.config = config
    
    def backtest(self, df: pd.DataFrame) -> dict:
        """
        Führt Backtest mit realistischen Annahmen durch
        """
        df = self.generate_signals(df)
        
        capital = self.config.initial_capital
        position = 0
        trades = []
        
        # Track Einstiegszeitpunkt für Look-ahead Vermeidung
        last_signal_time = None
        
        for idx, row in df.iterrows():
            # Signale werden erst im NÄCHSTEN Zeitraum umgesetzt
            # (verhindert Look-ahead Bias)
            if last_signal_time is not None:
                current_time = row['timestamp']
                time_diff = current_time - last_signal_time
                
                if last_signal == 'BUY' and position == 0:
                    # Kaufordersimulation mit Gebühren
                    effective_price = row['price'] * (1 + self.config.slippage_pct/100)
                    fee = effective_price * self.config.transaction_fee_pct / 100
                    total_cost = effective_price + fee
                    
                    if capital >= total_cost:
                        position = capital / total_cost
                        capital = 0
                        trades.append({
                            'type': 'BUY',
                            'price': effective_price,
                            'fee': fee,
                            'timestamp': current_time,
                            'capital': capital
                        })
                
                elif last_signal == 'SELL' and position > 0:
                    # Verkaufordersimulation
                    effective_price = row['price'] * (1 - self.config.slippage_pct/100)
                    fee = effective_price * self.config.transaction_fee_pct / 100
                    net_proceeds = (position * effective_price) - fee
                    
                    capital = net_proceeds
                    position = 0
                    trades.append({
                        'type': 'SELL',
                        'price': effective_price,
                        'fee': fee,
                        'timestamp': current_time,
                        'capital': capital
                    })
            
            # Aktuelles Signal

Verwandte Ressourcen
Verwandte Artikel