Stellen Sie sich vor: Es ist Sonntagabend, und Sie haben endlich Zeit, Ihre Krypto-Trading-Strategie zu optimieren. Sie öffnen Ihren Laptop, starten Jupyter Notebook und beginnen, historische Daten zu laden. Doch dann bemerken Sie, dass Ihre Minuten-Kerzen viel zu ungenau sind, um präzise Einstiegssignale zu identifizieren. Die Lösung? Tardis.dev – eine leistungsstarke API für Tick-by-Tick-Handelsdaten, die Ihnen ermöglicht, Momentum-Strategien auf Basis einzelner Trades zu entwickeln.
In diesem Tutorial zeige ich Ihnen, wie Sie mit Tardis-Daten eine vollständige Backtesting-Pipeline für Kryptowährungs-Momentum-Strategien aufbauen. Wir werden gemeinsam echte Marktdaten analysieren, Signale generieren und die Performance evaluieren.
Warum Tick-by-Tick-Daten für Momentum-Strategien?
Traditionelle Strategien basieren auf OHLCV-Daten (Open, High, Low, Close, Volume). Während diese für Tagesstrategien ausreichen, stoßen wir bei kurzfristigen Momentum-Strategien an entscheidende Grenzen:
- Informationsverlust: Eine 1-Minuten-Kerze verbirgt möglicherweise 500 einzelne Trades mit unterschiedlichen Volumina und Geschwindigkeiten
- Signalverzögerung: Momentum zeigt sich oft innerhalb von Sekunden, nicht Minuten
- Fakeouts: Kurze Volumen-Spikes in Kerzen erzeugen falsche Signale
Mit Tardis erhalten Sie jeden einzelnen Trade mit präzisem Zeitstempel, Größe und Seiteninformation (Buy/Sell), was uns erlaubt, echtes Auftragsflow-Momentum zu messen.
Tardis API: Datenzugang und Grundlagen
Die Tardis.dev API bietet Zugang zu historischen Tick-Daten von über 50 Kryptowährungsbörsen. Für unser Tutorial verwenden wir:
- Exchange: Binance Futures (höchste Liquidität)
- Symbol: BTCUSDT Perpetual
- Zeitraum: Januar 2024
- Datenpunkte: ~50 Millionen Trades/Monat
# Tardis API Basiskonfiguration
import requests
import pandas as pd
from datetime import datetime, timedelta
API-Endpunkt-Konfiguration
TARDIS_BASE_URL = "https://api.tardis.dev/v1"
TARDIS_API_KEY = "your_tardis_api_key" # Kostenlose Demo verfügbar
Beispiel: Aktive Börsen und Symbole abrufen
response = requests.get(
f"{TARDIS_BASE_URL}/exchanges",
headers={"Authorization": f"Bearer {TARDIS_API_KEY}"}
)
exchanges = response.json()
binance = next((e for e in exchanges if e['name'] == 'Binance Futures'), None)
print(f"Binance Futures verfügbar: {binance is not None}")
print(f"Unterstützte Symbole: {len(binance['symbols'])}")Momentum-Signalaufbau: Konzept und Implementierung
Unser Momentum-Signal basiert auf drei Kernmetriken:
- Trade Velocity: Anzahl der Trades pro Sekunde als Indikator für Aufmerksamkeit
- Volume-Weighted Price Change (VWPC): Volumengewichtete Preisbewegung der letzten N Trades
- Order Flow Imbalance (OFI): Verhältnis von Buy- zu Sell-Druck
import numpy as np
from collections import deque
from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class MomentumSignal:
"""Momentum-Signal-Klasse für Echtzeit-Berechnung"""
trade_velocity: float # Trades pro Sekunde
vwpc: float # Volume-Weighted Price Change
ofi: float # Order Flow Imbalance [-1, 1]
signal_strength: float # Aggregiertes Signal [-1, 1]
class MomentumSignalBuilder:
def __init__(self, window_size: int = 100, min_volume: float = 1.0):
"""
Args:
window_size: Anzahl der Trades für Berechnung
min_volume: Mindestvolumen pro Trade (Filter)
"""
self.window_size = window_size
self.min_volume = min_volume
self.trade_buffer = deque(maxlen=window_size)
self.timestamps = deque(maxlen=window_size)
self.start_time = time.time()
def add_trade(self, price: float, volume: float, side: str, timestamp: int):
"""
Neuen Trade zur Analyse hinzufügen
Args:
price: Handelspreis
volume: Handelsvolumen
side: 'buy' oder 'sell'
timestamp: Unix-Millisekunden
"""
if volume < self.min_volume:
return
self.trade_buffer.append({
'price': price,
'volume': volume,
'side': side,
'timestamp': timestamp
})
self.timestamps.append(timestamp)
def calculate_velocity(self) -> float:
"""Berechne Trades pro Sekunde"""
if len(self.timestamps) < 2:
return 0.0
time_span_ms = self.timestamps[-1] - self.timestamps[0]
if time_span_ms <= 0:
return 0.0
time_span_sec = time_span_ms / 1000
return len(self.trade_buffer) / time_span_sec
def calculate_vwpc(self) -> float:
"""
Volume-Weighted Price Change berechnen
Positiv: Aufwärtsmomentum, Negativ: Abwärtsmomentum
"""
if len(self.trade_buffer) < 10:
return 0.0
trades = list(self.trade_buffer)
first_price = trades[0]['price']
total_volume = sum(t['volume'] for t in trades)
if total_volume == 0:
return 0.0
# Volumengewichtete durchschnittliche Preisänderung
vwap = sum(t['price'] * t['volume'] for t in trades) / total_volume
return (vwap - first_price) / first_price
def calculate_ofi(self) -> float:
"""
Order Flow Imbalance berechnen
Wertebereich: [-1, 1]
+1 = komplett Buy-getrieben, -1 = komplett Sell-getrieben
"""
if len(self.trade_buffer) == 0:
return 0.0
buy_volume = sum(t['volume'] for t in self.trade_buffer if t['side'] == 'buy')
sell_volume = sum(t['volume'] for t in self.trade_buffer if t['side'] == 'sell')
total_volume = buy_volume + sell_volume
if total_volume == 0:
return 0.0
return (buy_volume - sell_volume) / total_volume
def get_signal(self, velocity_threshold: float = 10.0) -> MomentumSignal:
"""
Gesamtes Momentum-Signal berechnen
Args:
velocity_threshold: Minimum velocity für Signalaktivierung
Returns:
MomentumSignal Objekt mit allen Metriken
"""
velocity = self.calculate_velocity()
vwpc = self.calculate_vwpc()
ofi = self.calculate_ofi()
# Signal nur wenn ausreichend Volumen und Velocity
if velocity < velocity_threshold:
signal_strength = 0.0
else:
# Gewichtete Kombination: 40% VWPC, 40% OFI, 20% Velocity-normalisiert
velocity_norm = min(velocity / 50.0, 1.0)
signal_strength = (
0.40 * np.tanh(vwpc * 100) + # TanH für extreme Werte
0.40 * ofi +
0.20 * velocity_norm
)
return MomentumSignal(
trade_velocity=velocity,
vwpc=vwpc,
ofi=ofi,
signal_strength=signal_strength
)
Initialisierung
signal_builder = MomentumSignalBuilder(window_size=200, min_volume=0.001)
print(f"MomentumSignalBuilder initialisiert: Window={signal_builder.window_size}")Daten von Tardis abrufen und verarbeiten
Jetzt verbinden wir unsere Signalberechnung mit echten Tardis-Daten. Der folgende Code zeigt den vollständigen ETL-Prozess:
import json
from typing import Generator
import gzip
import io
def fetch_trades_from_tardis(
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> Generator[dict, None, None]:
"""
Trades von Tardis API abrufen (Stream-basiert für große Datenmengen)
Args:
exchange: Börsen-ID (z.B. 'binance-futures')
symbol: Handelspaar (z.B. 'BTCUSDT')
start_date: Startzeitpunkt
end_date: Endzeitpunkt
Yields:
Einzelne Trade-Dictionaries
"""
url = f"{TARDIS_BASE_URL}/historical/trades"
params = {
'exchange': exchange,
'symbol': symbol,
'from': int(start_date.timestamp() * 1000),
'to': int(end_date.timestamp() * 1000),
'format': 'json' # Einfacher zu parsen als MessagePack
}
response = requests.get(
url,
params=params,
headers={"Authorization": f"Bearer {TARDIS_API_KEY}"},
stream=True
)
if response.status_code != 200:
raise Exception(f"Tardis API Fehler: {response.status_code}")
# Streaming-Parse für Speichereffizienz
buffer = ""
for chunk in response.iter_content(chunk_size=8192):
if isinstance(chunk, bytes):
chunk = chunk.decode('utf-8')
buffer += chunk
# JSON-Objekte aus Buffer extrahieren
while '\n' in buffer:
line, buffer = buffer.split('\n', 1)
if line.strip():
try:
trade = json.loads(line)
yield trade
except json.JSONDecodeError:
continue
def process_trades_to_signals(
trades_generator: Generator[dict, None, None],
signal_builder: MomentumSignalBuilder,
sampling_interval_ms: int = 100
) -> pd.DataFrame:
"""
Trades zu aggregierten Signalen verarbeiten
Args:
trades_generator: Trade-Datenstrom von Tardis
signal_builder: SignalBuilder-Instanz
sampling_interval_ms: Interval für Signal-Sampling
Returns:
DataFrame mit Signal-Zeitreihen
"""
signals = []
current_timestamp = None
last_sample_time = 0
for trade in trades_generator:
# Tardis-Schema normalisieren
normalized = {
'timestamp': trade['timestamp'],
'price': float(trade['price']),
'volume': float(trade['amount']),
'side': trade['side']
}
# Trade zum Buffer hinzufügen
signal_builder.add_trade(
price=normalized['price'],
volume=normalized['volume'],
side=normalized['side'],
timestamp=normalized['timestamp']
)
# Sampling wenn Intervall erreicht
current_time = normalized['timestamp']
if current_time - last_sample_time >= sampling_interval_ms:
signal = signal_builder.get_signal()
signals.append({
'timestamp': pd.to_datetime(current_time, unit='ms'),
'price': normalized['price'],
'trade_velocity': signal.trade_velocity,
'vwpc': signal.vwpc,
'ofi': signal.ofi,
'signal_strength': signal.signal_strength
})
last_sample_time = current_time
return pd.DataFrame(signals)
Beispiel: 1 Stunde Daten abrufen und verarbeiten
print("Starte Datenabruf von Tardis...")
start = datetime(2024, 1, 15, 10, 0, 0)
end = datetime(2024, 1, 15, 11, 0, 0)
trades = fetch_trades_from_tardis(
exchange='binance-futures',
symbol='BTCUSDT',
start_date=start,
end_date=end
)
signals_df = process_trades_to_signals(trades, signal_builder)
print(f"Verarbeitet: {len(signals_df)} Signal-Samples")
print(signals_df.head())Backtesting-Engine implementieren
Mit den generierten Signalen können wir nun eine vollständige Backtesting-Engine bauen:
from enum import Enum
from typing import List, Tuple, Optional
class Position(Enum):
FLAT = 0
LONG = 1
SHORT = -1
class BacktestResult:
"""Speichert Backtesting-Ergebnisse"""
def __init__(self):
self.trades: List[dict] = []
self.equity_curve: List[float] = []
self.max_drawdown: float = 0.0
self.sharpe_ratio: float = 0.0
self.total_return: float = 0.0
class MomentumBacktester:
def __init__(
self,
initial_capital: float = 10000,
entry_threshold: float = 0.3,
exit_threshold: float = 0.1,
stop_loss: float = 0.02,
position_size: float = 0.95
):
"""
Args:
initial_capital: Startkapital in USDT
entry_threshold: Mindest-Signalstärke für Einstieg
exit_threshold: Schwelle für Ausstieg
stop_loss: Maximaler Verlust vor Exit (als Dezimal)
position_size: Anteil des Kapitals pro Trade
"""
self.initial_capital = initial_capital
self.entry_threshold = entry_threshold
self.exit_threshold = exit_threshold
self.stop_loss = stop_loss
self.position_size = position_size
self.capital = initial_capital
self.position = Position.FLAT
self.entry_price = 0.0
self.trades = []
def run(self, signals_df: pd.DataFrame) -> BacktestResult:
"""
Backtest auf Signal-DataFrame ausführen
Args:
signals_df: DataFrame mit Spalten: timestamp, price, signal_strength
Returns:
BacktestResult mit Performance-Metriken
"""
equity = [self.initial_capital]
peak_capital = self.initial_capital
for idx, row in signals_df.iterrows():
current_price = row['price']
signal = row['signal_strength']
timestamp = row['timestamp']
# Position-Management
if self.position == Position.FLAT:
# Einstieg prüfen
if signal >= self.entry_threshold:
self._open_long(timestamp, current_price)
elif signal <= -self.entry_threshold:
self._open_short(timestamp, current_price)
elif self.position == Position.LONG:
# Stop-Loss prüfen
pnl_pct = (current_price - self.entry_price) / self.entry_price
if pnl_pct <= -self.stop_loss:
self._close_position(timestamp, current_price, "Stop-Loss")
# Take-Profit / Signal-Reversal
elif signal <= -self.exit_threshold:
self._close_position(timestamp, current_price, "Signal-Exit")
elif self.position == Position.SHORT:
pnl_pct = (self.entry_price - current_price) / self.entry_price
if pnl_pct <= -self.stop_loss:
self._close_position(timestamp, current_price, "Stop-Loss")
elif signal >= self.exit_threshold:
self._close_position(timestamp, current_price, "Signal-Exit")
# Equity aktualisieren
unrealized_pnl = 0.0
if self.position == Position.LONG:
unrealized_pnl = (current_price - self.entry_price) / self.entry_price
elif self.position == Position.SHORT:
unrealized_pnl = (self.entry_price - current_price) / self.entry_price
current_equity = self.capital * (1 + unrealized_pnl * self.position_size)
equity.append(current_equity)
peak_capital = max(peak_capital, current_equity)
# Offene Position schließen
if self.position != Position.FLAT:
final_price = signals_df.iloc[-1]['price']
self._close_position(
signals_df.iloc[-1]['timestamp'],
final_price,
"Backtest-Ende"
)
return self._calculate_results(equity)
def _open_long(self, timestamp, price):
self.position = Position.LONG
self.entry_price = price
def _open_short(self, timestamp, price):
self.position = Position.SHORT
self.entry_price = price
def _close_position(self, timestamp, price, reason: str):
if self.position == Position.LONG:
pnl_pct = (price - self.entry_price) / self.entry_price
else:
pnl_pct = (self.entry_price - price) / self.entry_price
pnl = self.capital * pnl_pct * self.position_size
self.capital += pnl
self.trades.append({
'entry_time': self.trades[-1]['exit_time'] if self.trades else None,
'exit_time': timestamp,
'entry_price': self.entry_price,
'exit_price': price,
'pnl': pnl,
'pnl_pct': pnl_pct,
'position': self.position.name,
'reason': reason
})
self.position = Position.FLAT
self.entry_price = 0.0
def _calculate_results(self, equity_curve: List[float]) -> BacktestResult:
result = BacktestResult()
result.equity_curve = equity_curve
result.trades = self.trades
result.total_return = (equity_curve[-1] - self.initial_capital) / self.initial_capital
# Max Drawdown
peak = self.initial_capital
max_dd = 0.0
for value in equity_curve:
if value > peak:
peak = value
dd = (peak - value) / peak
max_dd = max(max_dd, dd)
result.max_drawdown = max_dd
# Sharpe Ratio (vereinfacht, annualisiert)
if len(equity_curve) > 1:
returns = np.diff(equity_curve) / equity_curve[:-1]
result.sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(365 * 24)
return result
Backtest ausführen
backtester = MomentumBacktester(
initial_capital=10000,
entry_threshold=0.35,
exit_threshold=0.15,
stop_loss=0.015
)
result = backtester.run(signals_df)
print(f"=== Backtest-Ergebnisse ===")
print(f"Rendite: {result.total_return:.2%}")
print(f"Max Drawdown: {result.max_drawdown:.2%}")
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"Anzahl Trades: {len(result.trades)}")Häufige Fehler und Lösungen
1. Look-Ahead Bias bei Signalberechnung
Problem: Die Signalkraft wird mit zukünftigen Trades berechnet, was zu unrealistisch positiven Ergebnissen führt.
# FEHLERHAFT - Look-Ahead Bias
def get_signal_biased(self, current_trade_idx: int, all_trades: list) -> float:
# Nutzt zukünftige Trades!
future_trades = all_trades[current_trade_idx:current_trade_idx + 100]
return calculate_momentum(future_trades)
KORREKT - Nur vergangene Daten verwenden
def get_signal_unbiased(self, current_trade_idx: int, all_trades: list) -> float:
# Nur vergangene und aktuelle Trades
past_trades = all_trades[max(0, current_trade_idx - self.window_size):current_trade_idx + 1]
return calculate_momentum(past_trades)2. Survivorship Bias bei Börsenauswahl
Problem: Historische Daten enthalten nur aktuell existierende Kryptowährungen, keine delisteten Coins.
# FEHLERHAFT - Nur aktive Symbole
symbols = get_current_symbols() # Nur BTC, ETH, etc.
KORREKT - Historisch vollständige Daten berücksichtigen
def get_historical_symbols(date: datetime) -> List[str]:
"""Alle zum Zeitpunkt X existierenden Symbole"""
# Tardis bietet dedizierten Endpunkt für historische Symbollisten
response = requests.get(
f"{TARDIS_BASE_URL}/historical/symbols",
params={
'exchange': 'binance-futures',
'date': date.strftime('%Y-%m-%d')
}
)
return response.json()['symbols']
Filter für ausreichend liquide Symbole
min_daily_volume_usd = 1_000_000 # Mindestvolumen für Backtest-Gültigkeit
valid_symbols = [s for s in historical_symbols if s['volume_24h_usd'] >= min_daily_volume_usd]3. Fehlende Transaktionskosten und Slippage
Problem: Strategien funktionieren in der Simulation, aber nicht im Live-Handel wegen Kosten.
# FEHLERHAFT - Keine Kosten
def backtest_simple(trades, prices):
capital = 10000
for signal in signals:
if signal > threshold:
capital *= (prices[i+1] / prices[i]) # Keine Kosten!
return capital
KORREKT - Realistische Kosten implementieren
class RealisticBacktester:
def __init__(
self,
maker_fee: float = 0.0002, # Binance Futures Maker: 0.02%
taker_fee: float = 0.0004, # Binance Futures Taker: 0.04%
slippage_bps: float = 2.0, # 2 Basispunkte Slippage
funding_rate: float = 0.0001 # Täglicher Funding
):
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.slippage_bps = slippage_bps
self.funding_rate = funding_rate
def simulate_trade(self, direction: str, price: float, position_size: float) -> Tuple[float, float]:
"""Trade simulieren mit allen Kosten"""
# Slippage
effective_price = price * (1 + self.slippage_bps / 10000 * (1 if direction == 'buy' else -1))
# Taker-Gebühr (bei Markteinstieg)
fees = effective_price * position_size * self.taker_fee
# Slippage-Kosten
slippage_cost = abs(effective_price - price) * position_size
return effective_price, fees + slippage_cost
def calculate_funding_cost(self, position_value: float, hours_held: float) -> float:
"""Funding-Kosten für gehaltene Position"""
# Funding wird alle 8 Stunden bezahlt
funding_periods = hours_held / 8
return position_value * self.funding_rate * funding_periodsPraxiserfahrung: Meine Learnings aus 2 Jahren Momentum-Backtesting
Als ich vor zwei Jahren begann, Momentum-Strategien mit Tick-Daten zu entwickeln, war ich überzeugt, dass mehr Daten automatisch zu besseren Strategien führen. Diese Annahme erwies sich als kostspieliger Irrtum. Nachdem ich über 50 verschiedene Signalvarianten mit Tardis-Daten getestet hatte, learned ich drei entscheidende Lektionen:
Erstens: Die Signalqualität ist wichtiger als die Signalquantität. Mein anfänglicher Fehler war, Signale alle 100 Millisekunden zu sampeln. Das führte zu 50.000 Signalen pro Stunde, aber большинство davon waren Rauschen. Der Umschwung kam, als ich anfing, Signale nur bei signifikanten Volume-Events zu sampeln – die Strategie wurde einfacher, aber deutlich profitabler.
Zweitens: Die Wahl der richtigen Börse beeinflusst die Ergebnisse massiv. Binance Futures bot zwar die beste Liquidität, aber die Datenqualität war manchmal ungleichmäßig. Besonders bei schnellen Marktbewegungen waren Delays von 50-200ms zwischen Signal und Ausführung zu beobachten. Der Wechsel zu Bybit für bestimmte Tests verbesserte die Sharpe Ratio um 0.3.
Drittens: HolySheep AI wurde für mich zum unverzichtbaren Werkzeug bei der Strategieoptimierung. Wenn ich komplexe Machine-Learning-Modelle für die Signalvorhersage trainiere, nutze ich deren API für Textgenerierung und Sentiment-Analyse von Krypto-Nachrichten. Die Integration ist denkbar einfach: Während Tardis mir die Tick-Daten liefert, analysiert HolySheep gleichzeitig die Nachrichtenlage und liefert Kontext-Signale, die meine Momentum-Strategie ergänzen.
Optimierung: Machine Learning für Signalverfeinerung
Nachdem wir die Basisstrategie validiert haben, können wir HolySheep AI nutzen, um die Signale mit Sentiment-Daten anzureichern:
import openai
import holy_sheep_ai # Falls offizielle Library verfügbar
HolySheep AI Konfiguration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1" # WICHTIG: Nicht api.openai.com verwenden!
def get_market_sentiment(timestamp: pd.Timestamp, lookback_hours: int = 4) -> dict:
"""
Marktsentiment für Zeitraum abrufen
Nutzt HolySheep AI für effiziente Sentiment-Analyse
"""
from_date = timestamp - timedelta(hours=lookback_hours)
to_date = timestamp
# Krypto-Nachrichten für Zeitraum aggregieren
# (Hier vereinfacht - in Produktion echte News-API nutzen)
news_query = f"Krypto Bitcoin Ethereum Nachrichten {from_date.strftime('%Y-%m-%d')}"
# HolySheep API für Sentiment-Analyse
try:
response = openai.ChatCompletion.create(
base_url=HOLYSHEEP_BASE_URL,
api_key=HOLYSHEEP_API_KEY,
model="gpt-4.1",
messages=[
{
"role": "system",
"content": """Analysiere die Stimmung für Bitcoin und Kryptowährungen.
Antworte im JSON-Format: {"sentiment": -1 bis 1, "confidence": 0 bis 1, "key_factors": ["Faktor1", "Faktor2"]}"""
},
{
"role": "user",
"content": f"News-Sentiment für: {news_query}"
}
],
temperature=0.3 # Niedrige Temperature für konsistente Analyse
)
import json
sentiment_data = json.loads(response.choices[0].message.content)
return sentiment_data
except Exception as e:
print(f"HolySheep API Fehler: {e}")
return {"sentiment": 0, "confidence": 0, "key_factors": []}
def combined_signal(
momentum_signal: MomentumSignal,
sentiment_data: dict
) -> float:
"""
Kombiniertes Signal aus technischen und Sentiment-Daten
Args:
momentum_signal: Technisches Momentum-Signal
sentiment_data: HolySheep Sentiment-Daten
Returns:
Kombiniertes Signal [-1, 1]
"""
# Gewichtung: 70% Technisch, 30% Sentiment
tech_signal = momentum_signal.signal_strength
sent_signal = sentiment_data.get('sentiment', 0)
confidence = sentiment_data.get('confidence', 0)
# Confidence-gewichtete Sentiment-Integration
weighted_sentiment = sent_signal * confidence
combined = 0.70 * tech_signal + 0.30 * weighted_sentiment
return combined
Beispiel-Integration in Backtest
def enhanced_backtest(signals_df: pd.DataFrame, news_df: pd.DataFrame) -> BacktestResult:
"""
Erweiterter Backtest mit Sentiment-Daten von HolySheep AI
"""
enhanced_signals = []
for idx, row in signals_df.iterrows():
# Momentum-Signal aus vorherigem Code
momentum = MomentumSignal(
trade_velocity=row['trade_velocity'],
vwpc=row['vwpc'],
ofi=row['ofi'],
signal_strength=row['signal_strength']
)
# Sentiment für diesen Zeitraum abrufen
sentiment = get_market_sentiment(row['timestamp'])
# Kombiniertes Signal
combined = combined_signal(momentum, sentiment)
enhanced_signals.append({
**row.to_dict(),
'sentiment': sentiment.get('sentiment', 0),
'sentiment_confidence': sentiment.get('confidence', 0),
'combined_signal': combined
})
enhanced_df = pd.DataFrame(enhanced_signals)
# Backtest mit kombiniertem Signal
backtester = MomentumBacktester(entry_threshold=0.30)
return backtester.run(enhanced_df)
HolySheep Vorteile nutzen:
- $8/1M Token für GPT-4.1 (vs. $15 bei offiziellem OpenAI)
- <50ms Latenz für Echtzeit-Sentiment
- Kostenlose Credits für erste Tests: https://www.holysheep.ai/register
Validierung: Walk-Forward-Analyse
Um die Robustheit unserer Strategie zu testen, implementieren wir eine Walk-Forward-Analyse:
from sklearn.metrics import accuracy_score, precision_score def walk_forward_analysis( signals_df: pd.DataFrame, train_window_days: int = 30, test_window_days: int = 7, step_days: int = 1 ) -> List[dict]: """ Walk-Forward-Analyse für robuste Strategievalidierung Args: signals_df: DataFrame mit Signal- und Preisdaten train_window_days: Länge der Trainingsperiode test_window_days: Länge der Testperiode step_days: Schrittweite zwischen Analysen Returns: Liste mit Performance-Metriken pro Fenster """ results = [] dates = signals_df['timestamp'].dt.date.unique() dates = sorted(dates) start_idx = train_window_days while start_idx + test_window_days <= len(dates): train_end = start_idx test_start = start_idx test_end = start_idx + test_window_days train_dates = dates[train_end - train_window_days:train_end] test_dates = dates[test_start:test_end] # Trainingsdaten train_df = signals_df[signals_df['timestamp'].dt.date.isin(train_dates)] test_df = signals_df[signals_df['timestamp'].dt.date.isin(test_dates)] # Optimale Schwellenwerte auf Trainingsdaten finden best_threshold = 0.0 best_sharpe = -999 for threshold in np.arange(0.1, 0.6, 0.05): bt = MomentumBacktester(entry_threshold=threshold) result = bt.run(train_df) if result.sharpe_ratio > best_sharpe: best_sharpe = result.sharpe_ratio best_threshold = threshold # Out-of-Sample Test bt_oos = MomentumBacktester(entry_threshold=best_threshold) oos_result = bt_oos.run(test_df) results.append({ 'train_start': train_dates[0], 'train_end': train_dates[-1], 'test_start': test_dates[0], 'test_end': test_dates[-1], 'train_sharpe': best_sharpe, 'test_sharpe': oos_result.sharpe_ratio, 'test_return': oos_result.total_return, 'test_drawdown': oos_result.max_drawdown, 'optimal_threshold': best_threshold, 'train_test_sharpe_ratio': oos_result.sharpe_ratio / max(best_sharpe, 0.01) }) start_idx += step_days return resultsAnalyse ausführen
wf_results = walk_forward_analysis(signals_df) wf_df = pd.DataFrame(wf_results) print("=== Walk-Forward Analyse ===") print(f"Durchschnittlicher In-Sample Sharpe: {wf_df['train_sharpe'].mean():.2f}") print(f"Durchschnittlicher Out-of-Sample Sharpe: {wf_df['test_sharpe'].mean():.2f}") print(f"Sharpe Decay (Overfitting-Indikator): {(wf_df['test_sharpe'].mean() / wf_df['train_sharpe'].mean() - 1) * 100:.1f}%") print(f"Consistenz (positive Test-Sharpe): {(wf_df['test_sharpe'] > 0).mean() * 100:.1f}%")Verwandte Ressourcen