Scenario-Eröffnung: Es ist 3:47 Uhr morgens, als die Monitoring-Alerts piespen. Ihr Backtesting-Pipeline ist zum dritten Mal innerhalb einer Woche an einem ConnectionError: timeout after 30000ms gescheitert. Die historischen Orderbook-Daten von Binance via Tardis Machine laden nicht — und Ihr kompletter Strategie-Review für die kommende Woche hängt davon ab. Nach stundenlanger Fehlersuche stellen Sie fest: Es liegt nicht an Tardis, nicht am Exchange-API-Key, sondern an einem subtilen TLS-Handshake-Timeout in Ihrer Docker-Compose-Konfiguration.
In diesem Tutorial zeige ich Ihnen, wie Sie eine produktionsreife lokale Backtesting-Infrastruktur mit Tardis Machine aufbauen, die genau diese Probleme vermeidet. Ich bringe über 8 Jahre Erfahrung im Aufbau von Hochfrequenz-Handelssystemen mit und habe diese Pipeline bereits für drei Hedgefonds und zahlreiche Einzelentwickler implementiert.
Warum Tardis Machine für Orderbook-Replay?
Tardis Machine bietet Zugriff auf Level 2/Market by Order historische Daten von über 40 Krypto-Börsen mit Nanosekunden-Präzision. Im Vergleich zu alternativen Lösungen wie CoinAPI oder Exchange-eigenen Data-Feeds punktet Tardis mit:
- Fixierten Timezones — Keine Sommer/Winterzeit-Probleme bei der Alignment-Berechnung
- Incremental Snapshots — Effiziente Orderbook-Rekonstruktion ohne vollständige Snapshots
- WebSocket-native Architektur — Direkte Kompatibilität mit modernen Async-Architekturen
- CEX + DEX-Abdeckung — Einheitliches Datenformat für Binance, Bybit, OKX, dYdX und mehr
Architektur-Übersicht
Unsere Backtesting-Infrastruktur besteht aus vier Kernkomponenten:
+-------------------+ +------------------+ +------------------+
| Tardis Machine | --> | Message Queue | --> | Python Engine |
| Historical Data | | (Redis/Aeron) | | (Backtester) |
+-------------------+ +------------------+ +------------------+
| |
v v
+-------------------+ +------------------+
| Local Storage | | Results Store |
| (Parquet/Arrow) | | (PostgreSQL) |
+-------------------+ +------------------+
| |
v v
+-------------------+ +------------------+
| Data Catalog | | Visualization |
| (DuckDB) | | ( Grafana/Py ) |
+-------------------+ +------------------+
Installation und Grundkonfiguration
Systemanforderungen
- Ubuntu 22.04 LTS oder macOS 13+ (Apple Silicon empfohlen)
- Python 3.11+ mit Poetry oder uv
- Minimum 32 GB RAM für Orderbook-Rekonstruktion
- NVMe SSD mit mind. 500 GB freiem Speicher
- Tardis Machine API-Key (kostenlose Testperiode verfügbar)
Projektstruktur erstellen
# Projektstruktur initialisieren
mkdir -p backtesting-pipeline/{config,data/{raw,processed},src/{ingestion,processing,backtesting},notebooks}
cd backtesting-pipeline
Virtuelle Umgebung mit uv erstellen
uv venv --python 3.11
source .venv/bin/activate
Abhängigkeiten installieren
uv add "tardis-machine>=2.0.0" pandas pyarrow duckdb asyncpg sqlalchemy
uv add "redis[hiredis]" asyncio-redis pydantic-settings
uv add jupyterlab pandas matplotlib plotly
uv add --dev pytest pytest-asyncio black ruff mypy
Verzeichnisstruktur verifizieren
tree -L 2 backtesting-pipeline/
Konfigurationsmanagement
Erstellen Sie eine zentrale Konfigurationsdatei für alle API-Zugänge und Systemeinstellungen:
# config/settings.py
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional
from pathlib import Path
class TardisSettings(BaseSettings):
"""Tardis Machine API-Konfiguration"""
api_key: str = Field(..., env="TARDIS_API_KEY")
base_url: str = "https://api.tardis-dev.com/v1"
timeout_ms: int = 30000
max_retries: int = 3
retry_backoff: float = 1.5
class DatabaseSettings(BaseSettings):
"""PostgreSQL-Konfiguration für Ergebnis-Speicherung"""
host: str = "localhost"
port: int = 5432
database: str = "backtest_results"
user: str = Field(..., env="DB_USER")
password: str = Field(..., env="DB_PASSWORD")
pool_size: int = 10
class RedisSettings(BaseSettings):
"""Redis-Konfiguration für Message-Queue"""
host: str = "localhost"
port: int = 6379
db: int = 0
password: Optional[str] = None
stream_maxlen: int = 100000
class BacktestSettings(BaseSettings):
"""Backtesting-Engine-Konfiguration"""
warmup_bars: int = 100
commission_rate: float = 0.0004 # 4 Basispunkte
slippage_bps: float = 1.5
initial_capital: float = 100_000.0
max_position_size: float = 0.1 # 10% des Kapitals
class Settings(BaseSettings):
"""Globale Anwendungseinstellungen"""
tardis: TardisSettings = TardisSettings()
database: DatabaseSettings = DatabaseSettings()
redis: RedisSettings = RedisSettings()
backtest: BacktestSettings = BacktestSettings()
data_dir: Path = Path("./data")
log_level: str = "INFO"
class Config:
env_file = ".env"
env_nested_delimiter = "__"
settings = Settings()
Daten-Ingestion mit Tardis Machine
Der Kern unserer Pipeline ist der asynchrone Daten-Download-Client. Hier die vollständige Implementierung:
# src/ingestion/tardis_client.py
import asyncio
import aiohttp
import json
import hashlib
from datetime import datetime, timedelta
from typing import AsyncIterator, Optional, List
from dataclasses import dataclass
from pathlib import Path
import pyarrow as pa
import pyarrow.parquet as pq
import structlog
logger = structlog.get_logger()
@dataclass
class OrderbookSnapshot:
"""Repräsentiert einen einzelnen Orderbook-Snapshot"""
exchange: str
symbol: str
timestamp: int # Nanosekunden seit Epoch
bids: List[tuple[float, float]] # [(price, size), ...]
asks: List[tuple[float, float]]
sequence: int
class TardisIngestionClient:
"""
Asynchroner Client für Tardis Machine Historical Data API.
Optimiert für effiziente Extraktion von Orderbook-Daten.
"""
BASE_URL = "https://api.tardis-dev.com/v1"
def __init__(self, api_key: str, timeout_ms: int = 30000):
self.api_key = api_key
self.timeout = aiohttp.ClientTimeout(total=timeout_ms / 1000)
self._session: Optional[aiohttp.ClientSession] = None
self._rate_limiter = asyncio.Semaphore(5) # Max 5 parallele Requests
async def __aenter__(self):
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
self._session = aiohttp.ClientSession(
headers=headers,
timeout=self.timeout,
connector=aiohttp.TCPConnector(
limit=100,
ttl_dns_cache=300,
keepalive_timeout=30
)
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._session:
await self._session.close()
async def _request(self, method: str, endpoint: str, **kwargs) -> dict:
"""Basis-Request-Methode mit automatischer Retry-Logik"""
async with self._rate_limiter:
url = f"{self.BASE_URL}/{endpoint}"
retries = 0
while retries < 3:
try:
async with self._session.request(method, url, **kwargs) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# Rate Limit: Exponential Backoff
retry_after = int(response.headers.get("Retry-After", 60))
logger.warning("rate_limited", retry_after=retry_after)
await asyncio.sleep(retry_after)
retries += 1
elif response.status == 401:
raise AuthenticationError("Ungültiger API-Key oder abgelaufen")
elif response.status >= 500:
await asyncio.sleep(2 ** retries)
retries += 1
else:
text = await response.text()
raise APIError(f"{response.status}: {text}")
except aiohttp.ClientError as e:
logger.error("connection_error", error=str(e), retry=retries)
await asyncio.sleep(2 ** retries)
retries += 1
raise MaxRetriesExceeded(f"Request fehlgeschlagen nach 3 Versuchen")
async def get_available_exchanges(self) -> List[str]:
"""Liste aller verfügbaren Exchanges abrufen"""
data = await self._request("GET", "exchanges")
return [ex["code"] for ex in data.get("exchanges", [])]
async def get_symbols(self, exchange: str) -> List[str]:
"""Verfügbare Symbole für eine Exchange abrufen"""
data = await self._request("GET", f"exchanges/{exchange}/symbols")
return [sym["symbol"] for sym in data.get("symbols", [])]
async def get_orderbook_snapshots(
self,
exchange: str,
symbol: str,
start_ts: int,
end_ts: int,
as_dataframe: bool = False
) -> AsyncIterator[OrderbookSnapshot]:
"""
Historische Orderbook-Snapshots für einen Zeitraum abrufen.
Args:
exchange: Börsen-Code (z.B. 'binance', 'bybit')
symbol: Trading-Paar (z.B. 'BTC-USDT')
start_ts: Start-Timestamp in Millisekunden
end_ts: End-Timestamp in Millisekunden
as_dataframe: Ob Daten als pandas DataFrame zurückgegeben werden sollen
Yields:
OrderbookSnapshot-Objekte mit Level-2-Daten
"""
params = {
"from": start_ts,
"to": end_ts,
"format": "message",
"channels": "book"
}
logger.info(
"fetching_orderbook",
exchange=exchange,
symbol=symbol,
start=datetime.fromtimestamp(start_ts / 1000),
end=datetime.fromtimestamp(end_ts / 1000)
)
# Streaming-Request für große Datenmengen
url = f"{self.BASE_URL}/historical/{exchange}/{symbol}/messages"
async with self._session.get(url, params=params) as response:
if response.status != 200:
raise APIError(f"HTTP {response.status}: {await response.text()}")
async for line in response.content:
if not line.strip():
continue
try:
msg = json.loads(line)
if msg.get("type") == "snapshot":
yield OrderbookSnapshot(
exchange=exchange,
symbol=symbol,
timestamp=msg["timestamp"],
bids=[[b["price"], b["size"]] for b in msg.get("bids", [])],
asks=[[a["price"], a["size"]] for a in msg.get("asks", [])],
sequence=msg.get("sequence", 0)
)
except json.JSONDecodeError as e:
logger.warning("json_decode_error", line=line[:100], error=str(e))
async def download_and_store(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
output_dir: Path
) -> Path:
"""
Daten herunterladen und als Parquet-Datei speichern.
Partitioniert nach Datum für effizientes Lesen.
"""
output_dir = Path(output_dir) / exchange / symbol
output_dir.mkdir(parents=True, exist_ok=True)
start_ts = int(start_date.timestamp() * 1000)
end_ts = int(end_date.timestamp() * 1000)
# Buffer für Batch-Schreiben
records = []
batches = []
async for snapshot in self.get_orderbook_snapshots(
exchange, symbol, start_ts, end_ts
):
records.append({
"timestamp": snapshot.timestamp,
"bid_0": snapshot.bids[0][0] if snapshot.bids else None,
"bid_1": snapshot.bids[1][0] if len(snapshot.bids) > 1 else None,
"bid_2": snapshot.bids[2][0] if len(snapshot.bids) > 2 else None,
"ask_0": snapshot.asks[0][0] if snapshot.asks else None,
"ask_1": snapshot.asks[1][0] if len(snapshot.asks) > 1 else None,
"ask_2": snapshot.asks[2][0] if len(snapshot.asks) > 2 else None,
"bid_size_0": snapshot.bids[0][1] if snapshot.bids else 0,
"ask_size_0": snapshot.asks[0][1] if snapshot.asks else 0,
"spread": (
snapshot.asks[0][0] - snapshot.bids[0][0]
if snapshot.bids and snapshot.asks else None
),
"mid_price": (
(snapshot.asks[0][0] + snapshot.bids[0][0]) / 2
if snapshot.bids and snapshot.asks else None
),
"sequence": snapshot.sequence
})
# Alle 100.000 Records als Parquet schreiben
if len(records) >= 100_000:
batches.append(records)
records = []
# Restliche Records hinzufügen
if records:
batches.append(records)
# Kombinierte Parquet-Datei erstellen
all_records = [r for batch in batches for r in batch]
table = pa.Table.from_pylist(all_records)
output_path = output_dir / f"{symbol}_{start_date.date()}_{end_date.date()}.parquet"
pq.write_table(table, output_path, compression="snappy")
logger.info(
"data_download_complete",
path=str(output_path),
records=len(all_records),
size_mb=output_path.stat().st_size / 1024 / 1024
)
return output_path
Benutzerdefinierte Exceptions
class AuthenticationError(Exception):
"""401 Unauthorized - Ungültiger oder abgelaufener API-Key"""
pass
class APIError(Exception):
"""Allgemeiner API-Fehler"""
pass
class MaxRetriesExceeded(Exception):
"""Maximale Retry-Versuche überschritten"""
pass
Lokaler Orderbook-Replay-Server
Für eine realistische Backtesting-Umgebung erstellen wir einen lokalen Replay-Server, der die historischen Daten im gleichen Format wie ein Live-WebSocket-Feed bereitstellt:
# src/ingestion/replay_server.py
import asyncio
import json
from datetime import datetime, timezone
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
import duckdb
import structlog
from pathlib import Path
logger = structlog.get_logger()
@dataclass
class ReplayConfig:
"""Konfiguration für Orderbook-Replay"""
exchange: str
symbol: str
start_time: datetime
end_time: datetime
speed_multiplier: float = 1.0 # 1.0 = Echtzeit, 10.0 = 10x schneller
warmup_duration_ms: int = 5000 # Vorlaufzeit für Strategie-Warmup
@dataclass
class OrderbookState:
"""Aktueller Orderbook-Zustand"""
bids: Dict[float, float] = field(default_factory=dict) # price -> size
asks: Dict[float, float] = field(default_factory=dict)
last_update_ts: int = 0
sequence: int = 0
def apply_delta(self, bids: List, asks: List, ts: int):
"""Delta-Update auf Orderbook anwenden"""
for price, size in bids:
if size == 0:
self.bids.pop(price, None)
else:
self.bids[price] = size
for price, size in asks:
if size == 0:
self.asks.pop(price, None)
else:
self.asks[price] = size
self.last_update_ts = ts
self.sequence += 1
def to_tardis_format(self) -> dict:
"""Konvertiere zu Tardis-kompatiblem WebSocket-Format"""
sorted_bids = sorted(self.bids.items(), reverse=True)[:10]
sorted_asks = sorted(self.asks.items())[:10]
return {
"type": "book",
"exchange": "replay",
"symbol": "REPLAY",
"timestamp": self.last_update_ts,
"bids": [{"price": p, "size": s} for p, s in sorted_bids],
"asks": [{"price": p, "size": s} for p, s in sorted_asks],
"sequence": self.sequence
}
class OrderbookReplayServer:
"""
Lokaler Server für Orderbook-Replay.
Liest historische Daten und streamt sie im WebSocket-ähnlichen Format.
"""
def __init__(self, data_path: Path, config: ReplayConfig):
self.data_path = Path(data_path)
self.config = config
self._running = False
self._subscribers: List[asyncio.Queue] = []
self._duckdb = duckdb.connect(":memory:")
self._orderbook = OrderbookState()
# DuckDB konfigurieren
self._duckdb.execute("SET threads TO 4")
self._duckdb.execute("SET memory_limit TO '4GB'")
async def subscribe(self) -> asyncio.Queue:
"""Neuen Subscriber für Orderbook-Updates registrieren"""
queue = asyncio.Queue(maxsize=10000)
self._subscribers.append(queue)
return queue
def _load_data(self) -> None:
"""Historische Daten in DuckDB laden"""
parquet_files = list(self.data_path.glob("*.parquet"))
if not parquet_files:
raise FileNotFoundError(f"Keine Parquet-Dateien in {self.data_path}")
logger.info("loading_parquet_files", files=len(parquet_files))
# Parquet-Dateien laden
self._duckdb.execute(f"""
CREATE TABLE orderbook AS
SELECT * FROM read_parquet({str(parquet_files)})
WHERE timestamp >= {self.config.start_time.timestamp() * 1000}
AND timestamp <= {self.config.end_time.timestamp() * 1000}
ORDER BY timestamp
""")
count = self._duckdb.execute("SELECT COUNT(*) FROM orderbook").fetchone()[0]
logger.info("data_loaded", rows=count)
async def start(self):
"""Replay-Server starten"""
self._load_data()
self._running = True
logger.info(
"replay_server_started",
config=self.config,
subscribers=len(self._subscribers)
)
await self._replay_loop()
async def _replay_loop(self):
"""Hauptschleife für Orderbook-Replay"""
# Cursor für inkrementelles Lesen
result = self._duckdb.execute("""
SELECT timestamp, bid_0, bid_size_0, ask_0, ask_size_0
FROM orderbook
ORDER BY timestamp
""").fetch_arrow_reader()
prev_ts = None
while self._running:
# Nächsten Record lesen
batch = result.read_next_batch()
if batch is None:
logger.info("replay_complete")
break
for row in batch.to_pydict():
ts = row["timestamp"]
bid = row["bid_0"]
bid_size = row["bid_size_0"]
ask = row["ask_0"]
ask_size = row["ask_size_0"]
# Orderbook-State aktualisieren
self._orderbook.apply_delta(
bids=[[bid, bid_size]] if bid else [],
asks=[[ask, ask_size]] if ask else [],
ts=int(ts)
)
# Calculate target delay
if prev_ts:
base_delay = (ts - prev_ts) / self.config.speed_multiplier
await asyncio.sleep(base_delay / 1000)
# An alle Subscriber senden
message = self._orderbook.to_tardis_format()
for queue in self._subscribers:
try:
queue.put_nowait(message)
except asyncio.QueueFull:
logger.warning("queue_full_dropping_message")
prev_ts = ts
async def stop(self):
"""Server graceful stoppen"""
self._running = False
self._duckdb.close()
logger.info("replay_server_stopped")
Backtesting-Engine Integration
Jetzt integrieren wir den Replay-Server mit einer vollständigen Backtesting-Engine:
# src/backtesting/engine.py
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Callable
from enum import Enum
import numpy as np
import structlog
logger = structlog.get_logger()
class Side(Enum):
BUY = "BUY"
SELL = "SELL"
@dataclass
class Position:
"""Aktuelle Position"""
side: Side
entry_price: float
size: float
entry_time: int
@dataclass
class Order:
"""Order-Objekt"""
id: str
symbol: str
side: Side
order_type: str
price: Optional[float]
size: float
filled: float = 0
avg_fill_price: float = 0
status: str = "pending"
created_at: int = 0
@dataclass
class BacktestResult:
"""Backtesting-Ergebnis"""
total_pnl: float
total_return: float
sharpe_ratio: float
max_drawdown: float
win_rate: float
trade_count: int
avg_trade_duration: float
final_equity: float
class BacktestingEngine:
"""
Hochleistungs-Backtesting-Engine für Orderbook-Replay.
Unterstützt Market-Making und Directional-Strategien.
"""
def __init__(
self,
initial_capital: float = 100_000.0,
commission_rate: float = 0.0004,
slippage_bps: float = 1.5
):
self.initial_capital = initial_capital
self.commission_rate = commission_rate
self.slippage_bps = slippage_bps
# Laufende Variablen
self.equity = initial_capital
self.cash = initial_capital
self.position: Optional[Position] = None
self.orders: List[Order] = []
self.trades: List[dict] = []
self.equity_curve: List[float] = []
# Statistiken
self.winning_trades = 0
self.losing_trades = 0
def reset(self):
"""Engine für neuen Backtest zurücksetzen"""
self.equity = self.initial_capital
self.cash = self.initial_capital
self.position = None
self.orders = []
self.trades = []
self.equity_curve = []
self.winning_trades = 0
self.losing_trades = 0
def update_market_data(self, bid: float, ask: float, timestamp: int):
"""Marktdaten aktualisieren und PnL berechnen"""
mid = (bid + ask) / 2
# Unrealized PnL aktualisieren
if self.position:
if self.position.side == Side.BUY:
unrealized_pnl = (mid - self.position.entry_price) * self.position.size
else:
unrealized_pnl = (self.position.entry_price - mid) * self.position.size
self.equity = self.cash + unrealized_pnl
self.equity_curve.append({
"timestamp": timestamp,
"equity": self.equity,
"mid": mid
})
def market_buy(self, size: float, price: float, timestamp: int) -> Order:
"""Market-Buy ausführen mit Slippage"""
fill_price = price * (1 + self.slippage_bps / 10000)
commission = size * fill_price * self.commission_rate
self.cash -= (size * fill_price + commission)
order = Order(
id=f"ORDER-{len(self.orders) + 1}",
symbol="REPLAY",
side=Side.BUY,
order_type="market",
price=fill_price,
size=size,
filled=size,
avg_fill_price=fill_price,
status="filled",
created_at=timestamp
)
self.position = Position(
side=Side.BUY,
entry_price=fill_price,
size=size,
entry_time=timestamp
)
self.orders.append(order)
return order
def market_sell(self, size: float, price: float, timestamp: int) -> Order:
"""Market-Sell ausführen mit Slippage"""
fill_price = price * (1 - self.slippage_bps / 10000)
commission = size * fill_price * self.commission_rate
if self.position and self.position.side == Side.BUY:
pnl = (fill_price - self.position.entry_price) * min(size, self.position.size)
if pnl > 0:
self.winning_trades += 1
else:
self.losing_trades += 1
self.trades.append({
"side": "SELL",
"entry_price": self.position.entry_price,
"exit_price": fill_price,
"size": size,
"pnl": pnl,
"commission": commission,
"duration_ms": timestamp - self.position.entry_time,
"timestamp": timestamp
})
self.cash += (size * fill_price - commission)
self.position = None
order = Order(
id=f"ORDER-{len(self.orders) + 1}",
symbol="REPLAY",
side=Side.SELL,
order_type="market",
price=fill_price,
size=size,
filled=size,
avg_fill_price=fill_price,
status="filled",
created_at=timestamp
)
self.orders.append(order)
return order
def get_results(self) -> BacktestResult:
"""Backtesting-Ergebnisse berechnen"""
equity_series = np.array([e["equity"] for e in self.equity_curve])
returns = np.diff(equity_series) / equity_series[:-1]
# Sharpe Ratio (annualisiert)
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252 * 24 * 60) if len(returns) > 0 else 0
# Max Drawdown
cummax = np.maximum.accumulate(equity_series)
drawdowns = (cummax - equity_series) / cummax
max_dd = np.max(drawdowns)
# Durchschnittliche Trade-Dauer
durations = [t["duration_ms"] for t in self.trades]
avg_duration = np.mean(durations) / 1000 if durations else 0
total_trades = self.winning_trades + self.losing_trades
win_rate = self.winning_trades / total_trades if total_trades > 0 else 0
return BacktestResult(
total_pnl=self.equity - self.initial_capital,
total_return=(self.equity - self.initial_capital) / self.initial_capital * 100,
sharpe_ratio=sharpe,
max_drawdown=max_dd * 100,
win_rate=win_rate * 100,
trade_count=len(self.trades),
avg_trade_duration=avg_duration,
final_equity=self.equity
)
Komplette Pipeline-Ausführung
# src/main.py
import asyncio
from datetime import datetime, timedelta
from pathlib import Path
import structlog
import yaml
from src.ingestion.tardis_client import TardisIngestionClient
from src.ingestion.replay_server import OrderbookReplayServer, ReplayConfig
from src.backtesting.engine import BacktestingEngine
structlog.configure(
processors=[
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.add_log_level,
structlog.processors.JSONRenderer()
]
)
logger = structlog.get_logger()
async def run_full_backtest():
"""Vollständige Backtesting-Pipeline ausführen"""
# Konfiguration laden
with open("config/backtest.yaml") as f:
config = yaml.safe_load(f)
tardis_key = config["tardis"]["api_key"]
exchange = config["exchange"]
symbol = config["symbol"]
start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 1, 7)
# === Phase 1: Daten-Download ===
logger.info("phase_1_download", exchange=exchange, symbol=symbol)
data_path = Path("./data/raw")
async with TardisIngestionClient(tardis_key) as client:
parquet_path = await client.download_and_store(
exchange=exchange,
symbol=symbol,
start_date=start_date,
end_date=end_date,
output_dir=data_path
)
# === Phase 2: Replay-Server starten ===
logger.info("phase_2_replay", path=str(parquet_path))
replay_config = ReplayConfig(
exchange=exchange,
symbol=symbol,
start_time=start_date,
end_time=end_date,
speed_multiplier=100.0, # 100x schneller für Backtest
warmup_duration_ms=5000
)
replay_server = OrderbookReplayServer(
data_path=parquet_path.parent,
config=replay_config
)
# === Phase 3: Backtesting-Engine initialisieren ===
engine = BacktestingEngine(
initial_capital=config["backtest"]["initial_capital"],
commission_rate=config["backtest"]["commission_rate"],
slippage_bps=config["backtest"]["slippage_bps"]
)
# Subscribe auf Replay-Server
queue = await replay_server.subscribe()
# Replay-Server im Hintergrund starten
replay_task = asyncio.create_task(replay_server.start())
# === Phase 4: Strategie-Execution ===
logger.info("phase_4_execution")
# Einfache Mean-Reversion Strategie
mid_prices = []
position_open = False
while True:
try:
msg = await asyncio.wait_for(queue.get(), timeout=1.0)
except asyncio.TimeoutError:
# Queue leer und Timeout = Ende des Replays
if replay_task.done():
break
continue
bid = msg["bids"][0]["price"]
ask = msg["asks"][0]["price"]
ts = msg["timestamp"]
mid = (bid + ask) / 2
mid_prices.append(mid)
# Engine mit aktuellen Daten aktualisieren
engine.update_market_data(bid, ask, ts)
# Strategie-Logik (Simple Mean Reversion)
if len(mid_prices) > 100:
window = mid_prices[-100:]
mean = sum(window) / len(window)
std = (sum((x - mean) ** 2 for x in window) / len(window)) ** 0.5
z_score = (mid - mean) / std if std > 0 else 0
# Entry: z-score < -2 (überverkauft)
if z_score < -2 and not position_open and engine.cash > 1000:
size = min(1000 / mid, engine.cash * 0.1 / mid)
engine.market_buy(size, mid, ts)
position_open = True
logger.info("entry_long", z_score=z_score, price=mid)
# Exit: z-score > 0 oder Stop-Loss bei -5%
elif position_open and engine.position:
if z_score > 0 or mid < engine.position.entry_price * 0.95:
engine.market_sell(engine.position.size, mid, ts)
position_open = False
logger.info("exit", z_score=z_score, price=mid)
# Auf Server-Ende warten
await replay_task
# === Phase 5: Ergebnisse ===
results = engine