Einleitung: Historische Orderbuch-Daten für Trading-Strategien
Historische L2-Orderbuchdaten sind das Fundament quantitativer Handelsstrategien, Market-Making-Algorithmen und Volumenprofil-Analysen. In diesem Tutorial zeige ich Ihnen Schritt für Schritt, wie Sie mit der Tardis.dev API Binance Futures Orderbuch-Snapshots für ein ganzes Jahr herunterladen und mit pandas für Backtesting und Analyse aufbereiten.
Die Tardis.dev API bietet Zugriff auf historische Marktroph-Daten von über 50 Börsen mit Millisekunden-Präzision. Für ein Jahr BTCUSDT Perpetual Orderbuch-Daten (1-Sekunden-Intervall) fallen ca. 2,4 GB an Daten an.
Umgebung vorbereiten
Installieren Sie die erforderlichen Python-Pakete:
# Virtuelle Umgebung erstellen und Pakete installieren
python -m venv tardis-env
source tardis-env/bin/activate # Windows: tardis-env\Scripts\activate
pip install requests pandas numpy matplotlib jupyter
pip install pandas-ta # Für technische IndikatorenDaten von Tardis.dev herunterladen
Die Tardis.dev API bietet einen einfachen REST-Endpunkt für historische Daten. Für Orderbuch-Snapshots verwenden wir den orderbooks-Stream:
import requests
import json
import time
from datetime import datetime, timedelta
import pandas as pd
from pathlib import Path
==============================
KONFIGURATION
==============================
BINANCE_FUTURES_SYMBOL = "BTCUSDT"
START_DATE = "2025-01-01"
END_DATE = "2025-01-31"
OUTPUT_DIR = Path("./orderbook_data")
OUTPUT_DIR.mkdir(exist_ok=True)
Tardis.dev API-Token (von https://tardis.dev abrufen)
TARDIS_API_TOKEN = "YOUR_TARDIS_TOKEN"
==============================
TARDIS.DEV API CLIENT
==============================
class TardisClient:
"""Client für den Download von Tardis.dev historischen Daten."""
BASE_URL = "https://api.tardis.ml/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}",
"Content-Type": "application/json"
})
def download_orderbook(
self,
exchange: str,
symbol: str,
date: str,
chunk_size: int = 10000
) -> list:
"""
Lädt Orderbuch-Snapshots für ein bestimmtes Datum herunter.
Args:
exchange: Börsen-Identifier (z.B. 'binance-futures')
symbol: Handelspaar (z.B. 'BTCUSDT')
date: Datum im Format 'YYYY-MM-DD'
chunk_size: Anzahl der Records pro Request
Returns:
Liste von Orderbuch-Snapshots
"""
url = f"{self.BASE_URL}/download"
params = {
"exchange": exchange,
"symbol": symbol,
"date": date,
"format": "json",
"filters": json.dumps([
{"type": "orderbook", "depth": 10} # Top 10 Bid/Ask
])
}
all_records = []
offset = 0
while True:
params["offset"] = offset
params["limit"] = chunk_size
response = self.session.get(url, params=params)
response.raise_for_status()
data = response.json()
if not data:
break
all_records.extend(data)
print(f" {date}: {len(all_records)} Records heruntergeladen...")
if len(data) < chunk_size:
break
offset += chunk_size
time.sleep(0.1) # Rate Limiting respektieren
return all_records
def get_orderbook_snapshot(self, exchange: str, symbol: str, timestamp: int) -> dict:
"""
Ruft einen einzelnen Orderbuch-Snapshot für einen bestimmten Zeitstempel ab.
Nützlich für Live-Testing und Debugging.
"""
url = f"{self.BASE_URL}/replay"
params = {
"exchange": exchange,
"symbol": symbol,
"from": timestamp,
"to": timestamp + 1000,
"format": "json"
}
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()
==============================
DOWNLOAD FUNKTION
==============================
def download_monthly_data(
client: TardisClient,
symbol: str,
year: int,
month: int
) -> pd.DataFrame:
"""Lädt alle Orderbuch-Daten für einen Monat herunter."""
exchange = "binance-futures"
all_data = []
# Anzahl der Tage im Monat berechnen
if month == 12:
next_month = datetime(year + 1, 1, 1)
else:
next_month = datetime(year, month + 1, 1)
current = datetime(year, month, 1)
while current < next_month:
date_str = current.strftime("%Y-%m-%d")
print(f"Lade Daten für {date_str}...")
try:
records = client.download_orderbook(
exchange=exchange,
symbol=symbol,
date=date_str
)
for record in records:
all_data.append({
"timestamp": pd.to_datetime(record["timestamp"], unit="ms"),
"symbol": record["symbol"],
"bids": record["data"]["bids"],
"asks": record["data"]["asks"],
"local_timestamp": pd.Timestamp.now()
})
except requests.exceptions.HTTPError as e:
print(f" Fehler: {e}")
if e.response.status_code == 429:
print(" Rate Limit erreicht, warte 60 Sekunden...")
time.sleep(60)
current += timedelta(days=1)
df = pd.DataFrame(all_data)
return df
==============================
HAUPTPROGRAMM
==============================
def main():
print("=" * 60)
print("TARDIS.DEV ORDERBUCH DOWNLOAD")
print("=" * 60)
# Client initialisieren
client = TardisClient(TARDIS_API_TOKEN)
# Daten herunterladen
print(f"\nStarte Download für {BINANCE_FUTURES_SYMBOL}...")
print(f"Zeitraum: {START_DATE} bis {END_DATE}\n")
df = download_monthly_data(
client,
symbol=BINANCE_FUTURES_SYMBOL,
year=2025,
month=1
)
# Daten speichern
output_file = OUTPUT_DIR / f"{BINANCE_FUTURES_SYMBOL}_orderbook_2025-01.parquet"
df.to_parquet(output_file, index=False)
print(f"\nDaten gespeichert: {output_file}")
print(f"Gesamtgröße: {output_file.stat().st_size / 1024 / 1024:.2f} MB")
print(f"Records: {len(df):,}")
return df
if __name__ == "__main__":
df = main()Orderbuch-Daten mit pandas analysieren
Nach dem Download analysieren wir die Daten, berechnen Spread, Orderbuch-Ungleichgewicht und Depth-Profile:
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from pathlib import Path
==============================
DATEN LADEN
==============================
DATA_FILE = Path("./orderbook_data/BTCUSDT_orderbook_2025-01.parquet")
print("Lade Orderbuch-Daten...")
df = pd.read_parquet(DATA_FILE)
df = df.set_index("timestamp").sort_index()
print(f"Geladen: {len(df):,} Records")
print(f"Zeitraum: {df.index.min()} bis {df.index.max()}")
print(f"Speicherbedarf: {df.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB")
==============================
ORDERBUCH-METRIKEN BERECHNEN
==============================
def calculate_orderbook_metrics(bids: list, asks: list) -> dict:
"""Berechnet Spread und Ungleichgewicht aus Orderbuch."""
if not bids or not asks:
return {"spread": np.nan, "mid_price": np.nan, "imbalance": np.nan}
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
spread = best_ask - best_bid
spread_pct = spread / ((best_ask + best_bid) / 2) * 100
mid_price = (best_ask + best_bid) / 2
# Bid/Ask Volume berechnen (Top 10 Level)
bid_volume = sum(float(b[1]) for b in bids[:10])
ask_volume = sum(float(a[1]) for a in asks[:10])
# Orderbuch-Ungleichgewicht: (BidVol - AskVol) / (BidVol + AskVol)
total_volume = bid_volume + ask_volume
imbalance = (bid_volume - ask_volume) / total_volume if total_volume > 0 else 0
return {
"best_bid": best_bid,
"best_ask": best_ask,
"spread": spread,
"spread_pct": spread_pct,
"mid_price": mid_price,
"bid_volume": bid_volume,
"ask_volume": ask_volume,
"imbalance": imbalance
}
Metriken für alle Records berechnen
print("\nBerechne Orderbuch-Metriken...")
metrics = df.apply(
lambda row: calculate_orderbook_metrics(row["bids"], row["asks"]),
axis=1,
result_type="expand"
)
df = pd.concat([df, metrics], axis=1)
==============================
STATISTISCHE ZUSAMMENFASSUNG
==============================
print("\n" + "=" * 60)
print("ORDERBUCH STATISTIK (BTCUSDT Perpetual)")
print("=" * 60)
stats = df[["spread", "spread_pct", "imbalance", "bid_volume", "ask_volume"]].describe()
print(stats.round(4))
print(f"\nDurchschnittlicher Spread: {df['spread_pct'].mean():.4f}%")
print(f"Spread Standardabweichung: {df['spread_pct'].std():.4f}%")
print(f"Max Ungleichgewicht: {df['imbalance'].abs().max():.4f}")
==============================
SPREAD-VERTEILUNG ANALYSIEREN
==============================
Nur Stunden-Level für die Visualisierung
df_hourly = df.resample("1H").agg({
"spread_pct": ["mean", "std", "min", "max"],
"imbalance": ["mean", 'std'],
"bid_volume": "mean",
"ask_volume": "mean"
}).dropna()
print(f"\nStündliche Resamples: {len(df_hourly)}")
==============================
VOLUMEN-PROFIL BERECHNEN
==============================
def calculate_volume_profile(df: pd.DataFrame, price_levels: int = 50) -> pd.DataFrame:
"""
Erstellt ein Volumen-Profil basierend auf aggregierten Orderbuch-Daten.
Zeigt, wie viel Volumen auf jedem Pre-Level vorhanden ist.
"""
# Sample alle 5 Minuten für Performance
sample = df.iloc[::300] # Alle 5 Minuten
all_bids = []
all_asks = []
for _, row in sample.iterrows():
if isinstance(row["bids"], list):
for level in row["bids"][:price_levels]:
all_bids.append({
"price": float(level[0]),
"volume": float(level[1])
})
if isinstance(row["asks"], list):
for level in row["asks"][:price_levels]:
all_asks.append({
"price": float(level[0]),
"volume": float(level[1])
})
bid_df = pd.DataFrame(all_bids)
ask_df = pd.DataFrame(all_asks)
return bid_df, ask_df
bid_profile, ask_profile = calculate_volume_profile(df)
print(f"\nVolumen-Profil (Top 50 Level):")
print(f" Bid-Volumen gesamt: {bid_profile['volume'].sum():.2f} BTC")
print(f" Ask-Volumen gesamt: {ask_profile['volume'].sum():.2f} BTC")
==============================
VISUALISIERUNG
==============================
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
1. Spread über Zeit
ax1 = axes[0, 0]
df["spread_pct"].resample("1min").mean().dropna().plot(ax=ax1, alpha=0.7)
ax1.set_title("Spread % über Zeit (1-Minuten-Aggregation)")
ax1.set_ylabel("Spread (%)")
ax1.set_xlabel("Zeit")
ax1.grid(True, alpha=0.3)
2. Orderbuch-Ungleichgewicht
ax2 = axes[0, 1]
df["imbalance"].resample("5min").mean().dropna().plot(ax=ax2, alpha=0.7, color="purple")
ax2.set_title("Orderbuch-Ungleichgewicht (5-Min-Aggregation)")
ax2.set_ylabel("Ungleichgewicht (-1 bis +1)")
ax2.set_xlabel("Zeit")
ax2.grid(True, alpha=0.3)
ax2.axhline(y=0, color="black", linestyle="--", alpha=0.5)
3. Volumen-Profil Histogramm
ax3 = axes[1, 0]
ax3.hist(bid_profile["price"], bins=100, weights=bid_profile["volume"],
alpha=0.6, label="Bids", color="green")
ax3.hist(ask_profile["price"], bins=100, weights=ask_profile["volume"],
alpha=0.6, label="Asks", color="red")
ax3.set_title("Volumen-Profil (Preis vs. Volumen)")
ax3.set_xlabel("Preis (USDT)")
ax3.set_ylabel("Kumulatives Volumen")
ax3.legend()
ax3.grid(True, alpha=0.3)
4. Spread-Verteilung
ax4 = axes[1, 1]
df["spread_pct"].hist(bins=100, ax=ax4, alpha=0.7, edgecolor="black")
ax4.set_title("Spread-Verteilung (%)")
ax4.set_xlabel("Spread (%)")
ax4.set_ylabel("Häufigkeit")
ax4.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("orderbook_analysis.png", dpi=150)
print("\nVisualisierung gespeichert: orderbook_analysis.png")
plt.show()Orderbuch für Strategie-Backtesting nutzen
Historische Orderbuch-Daten eignen sich hervorragend für das Backtesting von Market-Making- und Arbitrage-Strategien:
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import Optional
==============================
SIMPLES MARKET-MAKING BACKTEST
==============================
@dataclass
class Order:
"""Repräsentiert eine simulierte Order."""
timestamp: pd.Timestamp
side: str # 'buy' oder 'sell'
price: float
size: float
filled: bool = False
fill_price: Optional[float] = None
class MarketMakerBacktest:
"""
Simpler Market-Making Backtester basierend auf Orderbuch-Daten.
Strategie: Platziere Limit-Orders auf beiden Seiten mit festem Spread.
"""
def __init__(
self,
inventory_target: float = 0.0,
max_inventory: float = 1.0,
spread_bps: float = 5.0 # Spread in Basispunkten
):
self.inventory_target = inventory_target
self.max_inventory = max_inventory
self.spread_bps = spread_bps
self.inventory = 0.0
self.cash = 0.0
self.trades = []
self.equity_curve = []
def calculate_order_prices(self, mid_price: float) -> tuple:
"""Berechnet Bid- und Ask-Preise basierend auf Mid-Price."""
spread = mid_price * (self.spread_bps / 10000)
bid_price = mid_price - spread / 2
ask_price = mid_price + spread / 2
return bid_price, ask_price
def simulate_fill(self, order: Order, market_price: float) -> bool:
"""
Simuliert ob eine Order ausgeführt wird basierend auf Orderbuch-Dynamik.
Vereinfachte Logik:
- Buy-Orders werden gefüllt wenn Preis fällt
- Sell-Orders werden gefüllt wenn Preis steigt
"""
if order.side == "buy" and market_price <= order.price:
order.filled = True
order.fill_price = order.price
return True
elif order.side == "sell" and market_price >= order.price:
order.filled = True
order.fill_price = order.price
return True
return False
def run_backtest(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Führt den Backtest auf Orderbuch-Daten aus.
Args:
df: DataFrame mit 'mid_price' Spalte
Returns:
DataFrame mit Equity-Kurve
"""
df = df.copy()
df["equity"] = 0.0
df["inventory"] = 0.0
for i, (ts, row) in enumerate(df.iterrows()):
if pd.isna(row.get("mid_price")):
continue
mid_price = row["mid_price"]
bid_price, ask_price = self.calculate_order_prices(mid_price)
# Simuliere Market-Order Execution
# (Vereinfacht: kauf wenn imbalance > 0.3, verkaufe wenn < -0.3)
imbalance = row.get("imbalance", 0)
if imbalance > 0.3 and self.inventory > -self.max_inventory:
# Zu viele Bids = Preis wird fallen, wir kaufen
self.inventory += 0.1
self.cash -= bid_price * 0.1
elif imbalance < -0.3 and self.inventory < self.max_inventory:
# Zu viele Asks = Preis wird steigen, wir verkaufen
self.inventory -= 0.1
self.cash += ask_price * 0.1
# P&L berechnen
pnl = self.cash + self.inventory * mid_price
df.at[ts, "equity"] = pnl
df.at[ts, "inventory"] = self.inventory
self.equity_curve.append({"timestamp": ts, "equity": pnl})
return df.dropna(subset=["equity"])
def get_performance(self) -> dict:
"""Berechnet Performance-Metriken."""
equity = [t["equity"] for t in self.equity_curve]
if len(equity) < 2:
return {}
returns = np.diff(equity) / equity[:-1]
return {
"total_pnl": equity[-1] - equity[0],
"total_return_pct": (equity[-1] - equity[0]) / abs(equity[0]) * 100 if equity[0] != 0 else 0,
"sharpe_ratio": np.mean(returns) / np.std(returns) * np.sqrt(252 * 24 * 60) if np.std(returns) > 0 else 0,
"max_drawdown": self._calculate_max_drawdown(equity),
"final_inventory": self.inventory
}
def _calculate_max_drawdown(self, equity: list) -> float:
"""Berechnet Maximum Drawdown."""
peak = equity[0]
max_dd = 0
for value in equity:
if value > peak:
peak = value
dd = (peak - value) / peak if peak > 0 else 0
if dd > max_dd:
max_dd = dd
return max_dd * 100
==============================
BACKTEST AUSFÜHREN
==============================
print("=" * 60)
print("MARKET-MAKING BACKTEST")
print("=" * 60)
Daten laden und vorbereiten
df = pd.read_parquet("./orderbook_data/BTCUSDT_orderbook_2025-01.parquet")
df = df.set_index("timestamp").sort_index()
Fehlende Werte behandeln
df["mid_price"] = df["mid_price"].ffill().bfill()
Nur ein Tag für schnellen Test
test_data = df["2025-01-15":"2025-01-15"]
print(f"Testzeitraum: {test_data.index.min()} bis {test_data.index.max()}")
print(f"Records: {len(test_data):,}")
Backtest ausführen
mm = MarketMakerBacktest(
inventory_target=0.0,
max_inventory=2.0,
spread_bps=3.0 # 3 Basispunkte Spread
)
results = mm.run_backtest(test_data)
Performance ausgeben
perf = mm.get_performance()
print("\nBACKTEST ERGEBNISSE:")
print(f" Gesamt-P&L: ${perf['total_pnl']:.2f}")
print(f" Rendite: {perf['total_return_pct']:.2f}%")
print(f" Sharpe Ratio: {perf['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {perf['max_drawdown']:.2f}%")
print(f" Finale Inventar: {perf['final_inventory']:.4f} BTC")Kostenvergleich: Tardis.dev vs. alternatives Datenfeed
Historische Marktdaten können teuer werden. Hier ein Vergleich der führenden Anbieter für Orderbuch-Daten:
| Anbieter | Binance Futures L2 | 1 Jahr Kosten | Latenz | Archive |
|---|---|---|---|---|
| Tardis.dev | ✓ Verfügbar | €249/Monat (Pay-as-you-go ab €0.001/Record) | <100ms | Ab 2020 |
| QuantPlatform | ✓ Verfügbar | €499/Monat (Minimum) | <200ms | Ab 2019 |
| Exegy | ✓ Verfügbar | Ab $2.000/Monat | <50μs | Benutzerdefiniert |
| DIY (Node.js Script) | ✓ Verfügbar | Cloud-Kosten (~$50/Monat) | <500ms | Self-hosted |
Meine Empfehlung: Für die meisten Quant-Entwickler und Forscher ist Tardis.dev das beste Preis-Leistungs-Verhältnis. Die JSON-basierte API ist einfach zu integrieren, und das Pay-as-you-go-Modell ermöglicht kostengünstige Experimente.
Geeignet / Nicht geeignet für
Geeignet für:
- Akademische Forschung – Orderbuch-Mikrostruktur-Studien, Liquiditätsanalyse
- Algo-Trading-Backtesting – Market-Making, Arbitrage, Momentum-Strategien
- Machine Learning – Feature Engineering für Preisprognosen
- Risk Management – Historische Volatilitäts- und Korrelationsanalysen
- Startup-Prototyping – Schnelle MVP-Entwicklung ohne teure Datenabos
Nicht geeignet für:
- Ultra-Low-Latency Trading – Mikrosekunden-Präzision erfordert dedizierte Feed-Handler
- Live-Trading – Tardis.dev ist für historische Daten, nicht für Echtzeit-Streams
- Regulatory Compliance – Für einige Jurisdiktionen sind lizenzierte Datenfeeds erforderlich
- High-Frequency Trading – Sub-Sekunden-Daten nur mit spezialisierten Anbietern
Preise und ROI
Die Tardis.dev Preisstruktur bietet exzellente Flexibilität:
- Free Tier: 100.000 Records/Monat kostenlos
- Pay-as-you-go: Ab €0.001 pro Record
- Unlimited Plan: €249/Monat für unbegrenzte Downloads
ROI-Analyse für ein typisches Forschungsprojekt:
| Szenario | Kosten | Nutzen | ROI-Bewertung |
|---|---|---|---|
| Masterarbeit Backtest | €0 (Free Tier) | 1 Jahr Daten, Thesis | ★★★★★ |
| Startup MVP | €249/Monat | Alle Börsen, unbegrenzt | ★★★★☆ |
| Hedge Fund Research | €499/Monat | Multi-Asset, dedizierter Support | ★★★★☆ |
Häufige Fehler und Lösungen
1. Rate Limit erreicht (HTTP 429)
Problem: Zu viele Anfragen in kurzer Zeit führen zu Blockierung.
# FEHLERHAFTER CODE:
def download_all(client, dates):
results = []
for date in dates:
# Rate Limit Trigger!
data = client.download_orderbook(symbol="BTCUSDT", date=date)
results.extend(data)
return results
LÖSUNG: Exponential Backoff mit Rate Limiting
import time
import ratelimit
@ratelimit.sleep_and_retry
@ratelimit.limits(calls=100, period=60) # Max 100 Aufrufe pro Minute
def download_with_rate_limit(client, date):
"""Download mit automatischer Rate-Limit-Behandlung."""
max_retries = 5
base_delay = 1
for attempt in range(max_retries):
try:
data = client.download_orderbook(symbol="BTCUSDT", date=date)
return data
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
# Exponential Backoff
delay = base_delay * (2 ** attempt)
print(f"Rate Limit erreicht. Warte {delay}s...")
time.sleep(delay)
else:
raise
except requests.exceptions.RequestException as e:
print(f"Netzwerkfehler: {e}")
time.sleep(base_delay)
return [] # Fallback bei Fehlschlag2. MemoryError bei großen Datensätzen
Problem: 1 Jahr Orderbuch-Daten (~2.4 GB) passen nicht in den RAM.
# FEHLERHAFTER CODE:
df = pd.read_parquet("full_year_data.parquet") # MemoryError!
df["spread"] = df.apply(calculate_spread, axis=1) # Langsam
LÖSUNG: Chunked Processing mit pandas
def process_in_chunks(filepath, chunk_size=100000):
"""
Verarbeitet große Parquet-Dateien in Chunks.
Speichert Zwischenergebnisse in CSV für Memory-Effizienz.
"""
processed_chunks = []
# chunksize funktioniert bei Parquet nicht direkt
# Wir verwenden stattdessen pyarrow
import pyarrow.parquet as pq
parquet_file = pq.ParquetFile(filepath)
for batch in parquet_file.iter_batches(batch_size=chunk_size):
chunk_df = batch.to_pandas()
# Berechnungen durchführen
chunk_df["spread"] = chunk_df.apply(
lambda row: float(row["asks"][0][0]) - float(row["bids"][0][0])
if row["asks"] and row["bids"] else None,
axis=1
)
processed_chunks.append(chunk_df)
# Fortschritt ausgeben
print(f"Chunk {len(processed_chunks)} verarbeitet, "
f"{len(chunk_df):,} Records")
# Alle Chunks zusammenführen
return pd.concat(processed_chunks, ignore_index=True)
Alternative: Dask für Out-of-Core Processing
import dask.dataframe as dd
ddf = dd.read_parquet("full_year_data.parquet")
ddf["spread"] = ddf.apply(calculate_spread, axis=1).compute()
3. Falsche Zeitstempel-Konvertierung
Problem: Zeitstempel werden falsch interpretiert, Daten erscheinen in der falschen Zeitzone.
# FEHLERHAFTER CODE:
df["timestamp"] = pd.to_datetime(df["timestamp"]) # Annahme: ms
ODER
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="s") # Falsch für Binance
LÖSUNG: Explizite Konvertierung mit UTC
def parse_tardis_timestamp(ts_value):
"""
Parst Tardis.dev Zeitstempel korrekt.
Tardis verwendet Unix-Millisekunden.
"""
if isinstance(ts_value, (int, float)):
# Millisekunden seit Epoch
return pd.to_datetime(ts_value, unit="ms", utc=True)
elif isinstance(ts_value, str):
# ISO 8601 Format
return pd.to_datetime(ts_value, utc=True)
else:
return pd.NaT
Anwendung:
df["timestamp"] = df["timestamp"].apply(parse_tardis_timestamp)
df["timestamp"] = df["timestamp"].dt.tz_convert("Europe/Berlin") # Lokale Zeitzone
Verifikation
print(f"Erster Timestamp: {df['timestamp'].iloc[0]}")
print(f"Typ: {df['timestamp'].dtype}")
assert df["timestamp"].tz is not None, "Zeitzone muss gesetzt sein!"
Verwandte Ressourcen
Verwandte Artikel