Willkommen zu unserem technischen Deep-Dive für erfahrene Ingenieure und quantitative Trader. In diesem Tutorial zeige ich Ihnen, wie Sie HolySheep AI nutzen, um eine performante Anbindung an Tardis für Binance USDM Perpetual Futures zu realisieren – inklusive Mark Price, Index Price und Funding Rate historischer Daten für rigoroses Backtesting.
Architektur-Überblick: HolySheep + Tardis + Backtesting-Stack
Bevor wir in den Code eintauchen, möchte ich die Gesamtarchitektur skizzieren, die wir in diesem Tutorial aufbauen werden. Das System besteht aus drei Hauptkomponenten: (1) Tardis als Datenaggregator für Binance USDM Perpetual Futures, (2) HolySheep AI als Inference-Backend für Signalgenerierung und Strategieoptimierung, und (3) das Backtesting-Framework, das beide Komponenten orchestriert.
Die Herausforderung bei historischen Backtests mit Funding Rates liegt darin, dass Sie nicht nur Trade-Daten, sondern auch drei korrelierte Zeitreihen gleichzeitig verarbeiten müssen: Mark Price (der gehandelte Preis), Index Price (der zugrundeliegende Referenzpreis) und Funding Payments (die periodischen Cashflow-Transfers zwischen Long- und Short-Positionen). Ein fehlerhafter Umgang mit diesen Daten führt zu sogenanntem "look-ahead bias" oder inkorrekter Slippage-Berechnung.
┌─────────────────────────────────────────────────────────────────┐
│ Backtesting Framework │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Tardis API │───▶│ Data Store │───▶│ Strategy Engine │ │
│ │ (Historical)│ │ (Parquet) │ │ (HolySheep LLM) │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ Mark + Index + Time-Series Signal + Position │
│ Funding Streams Alignment Generation │
└─────────────────────────────────────────────────────────────────┘Warum HolySheep für Quantitative Forschung?
Bevor wir den Code schreiben, lassen Sie mich erklären, warum HolySheep AI eine herausragende Wahl für quantitative Forschungsprojekte ist. Die Plattform bietet Zugang zu führenden LLMs mit einer Kostenstruktur, die für Iterations-intensive Forschungsarbeit optimiert ist. DeepSeek V3.2 kostet beispielsweise nur $0.42 pro Million Token – das ist 95% günstiger als Claude Sonnet 4.5 bei $15/MTok. Bei einem typischen Backtesting-Projekt, das 10-50 Millionen Token für Prompt-Engineering und Signalgenerierung verbraucht, ergibt sich eine monatliche Ersparnis von mehreren hundert Dollar.
Die durchschnittliche Latenz liegt unter 50ms, was für synchrone Inferenz during des Backtests ausreichend ist. Für asynchrone Strategien, die auf aggregierte Signale warten, ist die Latenz sogar vernachlässigbar. Die Unterstützung für WeChat und Alipay macht das Onboarding für asiatische Trader besonders einfach, und das Startguthaben erlaubt sofortige Experimente ohne Kreditkarte.
HolySheep Preise und ROI-Analyse
| Modell | Preis pro Mio. Token | Latenz (P50) | Bestes Einsatzgebiet |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 | ~35ms | Kostensensitive Strategien, Bulk-Processing |
| Gemini 2.5 Flash | $2.50 | ~25ms | Schnelle Inferenz, Echtzeit-Signale |
| GPT-4.1 | $8.00 | ~45ms | Höchste Qualität, komplexe Muster |
| Claude Sonnet 4.5 | $15.00 | ~50ms | Nuancen-Reiche Analyse |
ROI-Rechnung für Quantitative Forschung
Angenommen, Sie führen 100 Iterationen Backtesting durch, jede mit 500K Token Input und 50K Token Output (für Signalanalyse und Report-Generierung). Mit HolySheep DeepSeek V3.2 kostet das:
- Input: 100 × 500K × $0.42/1M = $21.00
- Output: 100 × 50K × $1.68/1M (2x Rate) = $8.40
- Gesamt: $29.40 für 100 Backtest-Iterationen
Mit OpenAI direkt wären es $115.00 – eine Ersparnis von 75%. Bei 1000 Iterationen pro Monat (realistisch für aggressive Strategie-Optimierung) sparen Sie über $850 monatlich.
Geeignet / Nicht Geeignet für
Geeignet für:
- Langfristige Strategie-R&D mit hohem Iterationsbedarf
- Multi-Strategie-Portfolios, die verschiedene LLMs für verschiedene Signaltypen nutzen
- Teams mit Budget-Bewusstsein, die nicht für unnötige Latenz zahlen möchten
- Asiatische Trader, die WeChat/Alipay Zahlungen bevorzugen
- Prototyping und Experimentieren mit neuen Strategie-Ansätzen
Nicht geeignet für:
- Ultra-Low-Latency HFT-Anforderungen (<1ms), die dedizierte Hardware erfordern
- Regulatorisch regulierte Umgebungen mit spezifischen Compliance-Anforderungen
- Strategien, die ausschließlich Closed-Source-Modelle erfordern (obwohl HolySheep diese ebenfalls anbietet)
Voraussetzungen und Setup
Bevor wir mit dem Code beginnen, benötigen Sie folgende Voraussetzungen: (1) Ein HolySheep AI Konto mit API-Key, (2) Tardis Machine Zugang für Binance Historical Data, (3) Python 3.10+ mit den erforderlichen Bibliotheken. Ich empfehle die Verwendung von virtualenv oder conda für die Isolation der Abhängigkeiten.
# Installation der erforderlichen Pakete
pip install pandas numpy pyarrow httpx aiohttp asyncio
pip install tardis-client # Falls Sie Tardis direkt ansprechen
pip install holy-sheep-sdk # HolySheep Python Client
Umgebungsvariablen setzen
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export TARDIS_API_KEY="YOUR_TARDIS_API_KEY"Implementierung: Tardis Data Fetcher
Der erste Schritt ist das Sammeln der historischen Daten von Tardis. Für Binance USDM Perpetual Contracts benötigen wir Mark Price, Index Price und Funding Rate mit jeweils 8-Stunden-Intervallen (Funding wird alle 8 Stunden abgerechnet). Die folgende Klasse kapselt den Tardis-API-Zugriff und konvertiert die Daten in ein optimiertes Parquet-Format für schnelles Lesen während des Backtests.
import httpx
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import asyncio
class TardisDataFetcher:
"""
Fetches historical Mark+Index+Funding data from Tardis for Binance USDM Perpetuals.
Optimiert für Batch-Downloads und effiziente Speicherung als Parquet.
"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=10.0),
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
)
async def fetch_funding_rates(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> pd.DataFrame:
"""
Lädt Funding Rate History für einen Perpetual Contract.
Funding wird alle 8 Stunden gezahlt – 3 Events pro Tag.
"""
url = f"{self.BASE_URL}/fees/funding-rates"
params = {
"exchange": exchange,
"symbol": symbol,
"from": start_date.isoformat(),
"to": end_date.isoformat(),
"format": "dataframe"
}
headers = {"Authorization": f"Bearer {self.api_key}"}
response = await self.client.get(url, params=params, headers=headers)
response.raise_for_status()
df = pd.DataFrame(response.json())
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.set_index("timestamp").sort_index()
return df
async def fetch_mark_index_prices(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> Dict[str, pd.DataFrame]:
"""
Lädt Mark Price und Index Price History.
Beide werden von Tardis als separate Streams bereitgestellt.
"""
url = f"{self.BASE_URL}/derivatives/quote"
mark_df = await self._fetch_stream(
exchange, f"{symbol}:mark", start_date, end_date
)
index_df = await self._fetch_stream(
exchange, f"{symbol}:index", start_date, end_date
)
return {"mark": mark_df, "index": index_df}
async def _fetch_stream(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> pd.DataFrame:
"""Interner Helper für Stream-Downloads."""
url = f"{self.BASE_URL}/historical/{exchange}/{symbol}"
params = {
"from": int(start_date.timestamp()),
"to": int(end_date.timestamp()),
"format": "dataframe"
}
headers = {"Authorization": f"Bearer {self.api_key}"}
response = await self.client.get(url, params=params, headers=headers)
response.raise_for_status()
df = pd.DataFrame(response.json())
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.set_index("timestamp")
return df.sort_index()
def save_to_parquet(
self,
data: Dict[str, pd.DataFrame],
output_path: str
):
"""
Speichert alle Daten als partitioniertes Parquet für effizientes Lesen.
Partitioniert nach Jahr-Monat für schnellen Zeitreihen-Zugriff.
"""
for name, df in data.items():
if df is None or df.empty:
continue
df["year_month"] = df.index.strftime("%Y-%m")
for partition, partition_df in df.groupby("year_month"):
partition_df = partition_df.drop(columns=["year_month"])
partition_path = f"{output_path}/{name}/{partition}.parquet"
partition_df.to_parquet(
partition_path,
engine="pyarrow",
compression="snappy",
row_group_size=10000
)
async def close(self):
await self.client.aclose()
Benchmark: Download-Geschwindigkeit
async def benchmark_tardis_fetch():
"""
Benchmark: 1 Jahr historische Daten für BTCUSDM
Erwartete Dauer: 2-5 Minuten je nach Netzwerk
"""
fetcher = TardisDataFetcher(api_key="YOUR_TARDIS_API_KEY")
start = datetime(2025, 1, 1)
end = datetime(2025, 12, 31)
import time
t0 = time.time()
funding = await fetcher.fetch_funding_rates("binance-futures", "BTCUSDM", start, end)
prices = await fetcher.fetch_mark_index_prices("binance-futures", "BTCUSDM", start, end)
elapsed = time.time() - t0
print(f"Download abgeschlossen in {elapsed:.1f} Sekunden")
print(f"Funding Records: {len(funding):,}")
print(f"Mark Price Records: {len(prices['mark']):,}")
print(f"Index Price Records: {len(prices['index']):,}")
print(f"Durchsatz: {(len(funding) + len(prices['mark'])) / elapsed:.0f} Records/s")Implementierung: HolySheep Strategie-Engine
Jetzt kommen wir zum Kernstück: Die Integration von HolySheep AI für die Signalgenerierung. Die Strategie-Engine verwendet HolySheep, um komplexe Muster in den Mark-Index-Funding-Beziehungen zu erkennen und Trading-Signale zu generieren. Der Clou ist die Verwendung von DeepSeek V3.2 für kostengünstige Bulk-Analysen und GPT-4.1 für die finale Qualitätsprüfung kritischer Signale.
import httpx
import json
import asyncio
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
import pandas as pd
@dataclass
class TradingSignal:
timestamp: datetime
symbol: str
action: str # 'BUY', 'SELL', 'HOLD'
confidence: float
reasoning: str
metadata: Dict
class HolySheepStrategyEngine:
"""
Strategie-Engine für quantitative Signale unter Verwendung von HolySheep AI.
Unterstützt Multi-Modell-Routing für Kostenoptimierung.
"""
BASE_URL = "https://api.holysheep.ai/v1" # Korrekter Endpunkt
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0, connect=5.0),
limits=httpx.Limits(max_connections=20, max_keepalive_connections=10)
)
# Model-Kostenmapping für Budget-Tracking
self.model_costs = {
"deepseek-v3.2": {"input": 0.42, "output": 0.84}, # $/MTok
"gpt-4.1": {"input": 8.0, "output": 16.0},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
}
self.total_cost = 0.0
self.total_tokens = 0
def _build_analysis_prompt(
self,
mark_price: float,
index_price: float,
funding_rate: float,
funding_8h: float,
recent_volatility: float,
price_history: List[float],
position_summary: Dict
) -> str:
"""
Erstellt einen optimierten Prompt für die Mark-Index-Funding-Analyse.
Das Format ist kritisch für die Token-Effizienz.
"""
return f"""Analysiere folgendes Binance USD-M Futures Setup:
MARK DATA:
- Mark Price: ${mark_price:,.2f}
- Index Price: ${index_price:,.2f}
- Mark-Index Divergence: {((mark_price - index_price) / index_price * 100):.4f}%
FUNDING DATA:
- Current Funding Rate (8h): {funding_rate*100:.4f}%
- Projected Daily Funding Cost: ${funding_8h*3*100:.2f} per 1M notional
VOLATILITY:
- 24h Historical Volatility: {recent_volatility*100:.2f}%
- Recent Price Trend: {[f'${p:.2f}' for p in price_history[-5:]]}
POSITION:
- Current Position: {position_summary.get('side', 'NONE')} {position_summary.get('size', 0)} contracts
- Unrealized PnL: ${position_summary.get('pnl', 0):.2f}
ANALYSE AUFGABE:
1. Bewerte die Mark-Index Arbitrage-Möglichkeit
2. Beurteile ob Funding-Rate-Trend fortsetzt
3. Generiere klares Trading-Signal: BUY, SELL, oder HOLD
4. Setze Konfidenz-Level 0-100%
Antworte im JSON-Format:
{{"signal": "BUY|SELL|HOLD", "confidence": 0-100, "reasoning": "Kurze Begründung", "entry_zone": "{{price_range}}", "risk_level": "LOW|MEDIUM|HIGH"}}"""
async def analyze_market_state(
self,
market_data: Dict,
model: str = "deepseek-v3.2"
) -> TradingSignal:
"""
Analysiert den aktuellen Markt-State und generiert ein Trading-Signal.
"""
prompt = self._build_analysis_prompt(
mark_price=market_data["mark_price"],
index_price=market_data["index_price"],
funding_rate=market_data["funding_rate"],
funding_8h=market_data["funding_8h"],
recent_volatility=market_data["volatility"],
price_history=market_data["price_history"],
position_summary=market_data.get("position", {})
)
payload = {
"model": model,
"messages": [
{"role": "system", "content": "Du bist ein erfahrener Krypto-Quant-Analyst. Antworte präzise und datengetrieben."},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # Niedrig für konsistente Signale
"max_tokens": 500,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
t0 = asyncio.get_event_loop().time()
response = await self.client.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
headers=headers
)
latency_ms = (asyncio.get_event_loop().time() - t0) * 1000
if response.status_code != 200:
raise Exception(f"HolySheep API Error: {response.status_code} - {response.text}")
result = response.json()
usage = result.get("usage", {})
# Kostenberechnung
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
cost = self._calculate_cost(model, input_tokens, output_tokens)
self.total_cost += cost
self.total_tokens += input_tokens + output_tokens
# Parse Response
content = result["choices"][0]["message"]["content"]
signal_data = json.loads(content)
return TradingSignal(
timestamp=datetime.now(),
symbol=market_data["symbol"],
action=signal_data.get("signal", "HOLD"),
confidence=signal_data.get("confidence", 50),
reasoning=signal_data.get("reasoning", ""),
metadata={
"latency_ms": latency_ms,
"tokens_used": input_tokens + output_tokens,
"cost_usd": cost,
"model": model,
"entry_zone": signal_data.get("entry_zone", ""),
"risk_level": signal_data.get("risk_level", "MEDIUM")
}
)
def _calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Berechnet die Kosten basierend auf Token-Verbrauch."""
rates = self.model_costs.get(model, {"input": 1.0, "output": 2.0})
input_cost = (input_tokens / 1_000_000) * rates["input"]
output_cost = (output_tokens / 1_000_000) * rates["output"]
return input_cost + output_cost
async def batch_analyze(
self,
market_data_list: List[Dict],
model: str = "deepseek-v3.2",
concurrency: int = 5
) -> List[TradingSignal]:
"""
Führt parallele Analysen mit Concurrency-Limit durch.
Wichtig für effizientes Backtesting mit tausenden von Zeitpunkten.
"""
semaphore = asyncio.Semaphore(concurrency)
async def limited_analyze(data):
async with semaphore:
return await self.analyze_market_state(data, model)
tasks = [limited_analyze(data) for data in market_data_list]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter Exceptions
valid_results = [r for r in results if isinstance(r, TradingSignal)]
errors = [r for r in results if not isinstance(r, TradingSignal)]
if errors:
print(f"Warnung: {len(errors)} von {len(results)} Analysen fehlgeschlagen")
return valid_results
def get_cost_summary(self) -> Dict:
"""Gibt eine Zusammenfassung der Kosten zurück."""
return {
"total_cost_usd": round(self.total_cost, 4),
"total_tokens": self.total_tokens,
"avg_cost_per_1k": round(self.total_cost / (self.total_tokens / 1000), 4) if self.total_tokens > 0 else 0,
"cost_per_signal": round(self.total_cost / max(1, self.total_tokens // 500), 4)
}
async def close(self):
await self.client.aclose()
Benchmark: HolySheep Latenz und Kosten
async def benchmark_holy_sheep():
"""
Benchmark für HolySheep Strategie-Engine.
Erwartete Latenz: 35-50ms für DeepSeek V3.2
"""
engine = HolySheepStrategyEngine(api_key="YOUR_HOLYSHEEP_API_KEY")
# Simuliere typische Marktdaten
sample_data = {
"symbol": "BTCUSDM",
"mark_price": 67542.50,
"index_price": 67538.20,
"funding_rate": 0.0001,
"funding_8h": 0.01,
"volatility": 0.025,
"price_history": [67200, 67350, 67400, 67500, 67542.50],
"position": {"side": "LONG", "size": 1.0, "pnl": 125.50}
}
# Single Request Benchmark
latencies = []
for _ in range(20):
t0 = asyncio.get_event_loop().time()
signal = await engine.analyze_market_state(sample_data)
latencies.append((asyncio.get_event_loop().time() - t0) * 1000)
print(f"HolySheep Latenz Benchmark (DeepSeek V3.2):")
print(f" P50: {sorted(latencies)[10]:.1f}ms")
print(f" P95: {sorted(latencies)[19]:.1f}ms")
print(f" P99: {max(latencies):.1f}ms")
print(f"\nKostenzusammenfassung: {engine.get_cost_summary()}")Implementierung: Backtesting Framework
Jetzt kombinieren wir beide Komponenten im vollständigen Backtesting-Framework. Dieses Framework ist für Produktionseinsatz optimiert: Es verwendet chunked Processing für Speichereffizienz, hat eingebaute Progress-Tracking und berechnet alle relevanten Metriken inklusive Funding-Kosten.
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import asyncio
from concurrent.futures import ThreadPoolExecutor
import json
class BacktestEngine:
"""
Produktionsreifes Backtesting Framework für Binance USDM Perpetual Strategien.
Integriert Tardis-Daten mit HolySheep AI Signalgenerierung.
"""
def __init__(
self,
tardis_fetcher: 'TardisDataFetcher',
strategy_engine: 'HolySheepStrategyEngine',
initial_capital: float = 100_000.0,
position_size_pct: float = 0.10
):
self.tardis = tardis_fetcher
self.strategy = strategy_engine
self.initial_capital = initial_capital
self.position_size_pct = position_size_pct
# Backtest State
self.equity_curve = []
self.trades = []
self.funding_costs = []
self.signals = []
async def run_backtest(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
timeframe_minutes: int = 60,
signal_model: str = "deepseek-v3.2"
) -> Dict:
"""
Führt den vollständigen Backtest durch.
Args:
symbol: Trading Pair (z.B. 'BTCUSDM')
start_date: Start des Backtest-Zeitraums
end_date: Ende des Backtest-Zeitraums
timeframe_minutes: Granularität der Signalanalyse
signal_model: HolySheep Modell für Signale
"""
print(f"Starte Backtest: {symbol}")
print(f"Zeitraum: {start_date.date()} bis {end_date.date()}")
# Phase 1: Daten laden
print("\n[1/4] Lade historische Daten von Tardis...")
funding_df = await self.tardis.fetch_funding_rates(
"binance-futures", symbol, start_date, end_date
)
price_data = await self.tardis.fetch_mark_index_prices(
"binance-futures", symbol, start_date, end_date
)
# Phase 2: Daten alignen
print("[2/4] Aligniere Mark/Index/Funding Zeitreihen...")
aligned_data = self._align_timeframes(
mark_df=price_data["mark"],
index_df=price_data["index"],
funding_df=funding_df,
timeframe_minutes=timeframe_minutes
)
# Phase 3: Signalgenerierung
print(f"[3/4] Generiere Trading-Signale ({len(aligned_data)} Zeitpunkte)...")
market_states = self._prepare_market_states(aligned_data)
# Batch-Processing für Effizienz
signals = []
chunk_size = 50
for i in range(0, len(market_states), chunk_size):
chunk = market_states[i:i+chunk_size]
chunk_signals = await self.strategy.batch_analyze(
chunk,
model=signal_model,
concurrency=10
)
signals.extend(chunk_signals)
# Progress
progress = min(100, (i + chunk_size) / len(market_states) * 100)
print(f" Fortschritt: {progress:.1f}%", end="\r")
print(f"\n {len(signals)} Signale generiert")
# Phase 4: Simulation
print("[4/4] Führe Trading-Simulation durch...")
results = self._simulate_trading(signals, aligned_data, funding_df)
return results
def _align_timeframes(
self,
mark_df: pd.DataFrame,
index_df: pd.DataFrame,
funding_df: pd.DataFrame,
timeframe_minutes: int
) -> pd.DataFrame:
"""
Aligniert alle Zeitreihen auf den gewählten timeframe.
Funding wird interpoliert für nicht-Funding-Zeitpunkte.
"""
# Resample auf timeframe
mark_resampled = mark_df["price"].resample(f"{timeframe_minutes}T").last()
index_resampled = index_df["price"].resample(f"{timeframe_minutes}T").last()
# Funding auf Zeitraum ausdehnen (letzter bekannter Wert)
funding_extended = funding_df["rate"].resample(f"{timeframe_minutes}T").last()
funding_extended = funding_extended.ffill()
# Zusammenführen
aligned = pd.DataFrame({
"mark_price": mark_resampled,
"index_price": index_resampled,
"funding_rate": funding_extended
})
# Mark-Index Divergenz berechnen
aligned["divergence"] = (aligned["mark_price"] - aligned["index_price"]) / aligned["index_price"]
# Volatilität berechnen (Rolling 24 Perioden)
aligned["volatility"] = aligned["mark_price"].pct_change().rolling(24).std()
# Drop NaN
aligned = aligned.dropna()
return aligned
def _prepare_market_states(self, aligned_data: pd.DataFrame) -> List[Dict]:
"""Konvertiert alignierte Daten in Market-State-Dicts für HolySheep."""
market_states = []
for timestamp, row in aligned_data.iterrows():
# Price History für Kontext
idx = aligned_data.index.get_loc(timestamp)
price_history = aligned_data["mark_price"].iloc[max(0, idx-20):idx+1].tolist()
state = {
"timestamp": timestamp,
"symbol": "BTCUSDM",
"mark_price": row["mark_price"],
"index_price": row["index_price"],
"funding_rate": row["funding_rate"],
"funding_8h": row["funding_rate"] * row["mark_price"],
"volatility": row["volatility"] if pd.notna(row["volatility"]) else 0.02,
"price_history": price_history,
"position": self._get_current_position()
}
market_states.append(state)
return market_states
def _get_current_position(self) -> Dict:
"""Gibt aktuelle Position zurück (interner State)."""
if not self.equity_curve:
return {"side": "NONE", "size": 0, "pnl": 0}
last_trade = self.trades[-1] if self.trades else None
if last_trade and last_trade["exit_price"] is None:
return {
"side": last_trade["side"],
"size": last_trade["size"],
"pnl": self._calculate_unrealized_pnl(last_trade)
}
return {"side": "NONE", "size": 0, "pnl": 0}
def _simulate_trading(
self,
signals: List,
aligned_data: pd.DataFrame,
funding_df: pd.DataFrame
) -> Dict:
"""
Simuliert Trading basierend auf Signalen.
Berechnet PnL, Slippage, Funding-Kosten.
"""
capital = self.initial_capital
position = None
equity = capital
for i, signal in enumerate(signals):
timestamp = signal.timestamp
price = aligned_data.loc[aligned_data.index == timestamp, "mark_price"].values[0]
# Funding-Kosten akkumulieren
funding_cost = self._calculate_funding_cost(position, price, funding_df, timestamp)
self.funding_costs.append({"timestamp": timestamp, "cost": funding_cost})
# Signal ausführen
if signal.action == "BUY" and position is None:
size = (capital * self.position_size_pct) / price
position = {
"side": "LONG",
"entry_price": price,
"size": size,
"entry_time": timestamp
}
self.trades.append({**position, "exit_price": None})
elif signal.action == "SELL" and position is None:
size = (capital * self.position_size_pct) / price
position = {
"side": "SHORT",
"entry_price": price,
"size": size,
"entry_time": timestamp
}
self.trades.append({**position, "exit_price": None})
elif signal.action == "HOLD" and position is not None:
# Close bei umgekehrtem Signal
pnl = self._calculate_realized_pnl(position, price)
equity += pnl - funding_cost
position["exit_price"] = price
position = None
# Equity updaten
if position:
unrealized = self._calculate_unrealized_pnl(position, price)
self.equity_curve.append({
"timestamp": timestamp,
"equity": equity + unrealized,
"position_pnl": unrealized
})
else:
self.equity_curve.append({
"timestamp": timestamp,
"equity": equity - funding_cost,
"position_pnl": 0
})
# Ergebnisse aggregieren
return self._aggregate_results(equity, funding_df)
def _calculate_funding_cost(
self,
position: Optional[Dict],