In der Welt des algorithmischen Handels und der quantitativen Analyse sind präzise Marktdaten das Fundament jeder erfolgreichen Strategie. Die Migration von historischen Tick-Daten zwischen verschiedenen Datenanbietern stellt jedoch selbst erfahrene Entwickler vor erhebliche technische Herausforderungen. In diesem umfassenden Playbook zeige ich Ihnen, wie Sie eine reibungslose Datenmigration von Bybit und Binance über Tardis.dev zu HolySheep AI durchführen – inklusive bewährter Strategien für Breakpoint-Resume, Deduplizierung und Zeitstempel-Normalisierung.
Warum eine Migration erforderlich ist
Bevor wir in die technischen Details eintauchen, sollten wir verstehen, warum Teams überhaupt eine Migration in Betracht ziehen:
- Kostenexplosion bei Tardis.dev: Die Preise für historische Tick-Daten sind in den letzten Quartalen signifikant gestiegen, während die Datenqualität bei bestimmten Assets inkonsistent bleibt
- Latenzprobleme: Tardis.dev bietet keine garantierten Latenzgarantien unter 100ms, was für Hochfrequenzhändler kritisch ist
- API-Limits und Throttling: Strikte Rate-Limits erschweren die Bulk-Extraktion großer Datensätze
- Fehlende Multimarkt-Unterstützung: Wer gleichzeitig Bybit und Binance analysieren möchte, benötigt separate Integrationen
Die technischen Herausforderungen verstehen
1. Breakpoint-Resume (断点续传)
Bei der Extraktion großer Datenmengen über Monate oder Jahre hinweg ist es unvermeidlich, dass Verbindungen unterbrechen oder Timeouts auftreten. Ein robustes Resume-System ist daher essentiell:
#!/usr/bin/env python3
"""
Bybit/Binance Tick-Daten Migration mit Breakpoint-Resume
Unterstützt both synchrone und asynchrone Datenquellen
"""
import asyncio
import aiohttp
import json
import sqlite3
from datetime import datetime, timedelta
from typing import Optional, Dict, List
from dataclasses import dataclass, asdict
import hashlib
@dataclass
class MigrationCheckpoint:
"""Speichert den aktuellen Fortschritt der Migration"""
exchange: str
symbol: str
start_time: int
end_time: int
last_trade_id: str
total_records: int
checksum: str
last_updated: datetime
class TickDataMigrator:
def __init__(self, db_path: str = "migration_state.db"):
self.db_path = db_path
self.checkpoints: Dict[str, MigrationCheckpoint] = {}
self._init_database()
def _init_database(self):
"""Initialisiert die lokale SQLite-Datenbank für Checkpoints"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS checkpoints (
id INTEGER PRIMARY KEY AUTOINCREMENT,
exchange TEXT NOT NULL,
symbol TEXT NOT NULL,
start_time INTEGER NOT NULL,
end_time INTEGER NOT NULL,
last_trade_id TEXT,
total_records INTEGER DEFAULT 0,
checksum TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(exchange, symbol, start_time)
)
""")
conn.commit()
conn.close()
def _generate_checkpoint_key(self, exchange: str, symbol: str, start_time: int) -> str:
"""Generiert einen eindeutigen Schlüssel für den Checkpoint"""
raw = f"{exchange}:{symbol}:{start_time}"
return hashlib.md5(raw.encode()).hexdigest()
def save_checkpoint(self, checkpoint: MigrationCheckpoint):
"""Speichert den aktuellen Fortschritt persistent"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO checkpoints
(exchange, symbol, start_time, end_time, last_trade_id,
total_records, checksum, last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
checkpoint.exchange,
checkpoint.symbol,
checkpoint.start_time,
checkpoint.end_time,
checkpoint.last_trade_id,
checkpoint.total_records,
checkpoint.checksum,
checkpoint.last_updated
))
conn.commit()
conn.close()
print(f"✅ Checkpoint gespeichert: {checkpoint.exchange}/{checkpoint.symbol}")
def load_checkpoint(self, exchange: str, symbol: str,
start_time: int) -> Optional[MigrationCheckpoint]:
"""Lädt den letzten gespeicherten Fortschritt"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT exchange, symbol, start_time, end_time,
last_trade_id, total_records, checksum, last_updated
FROM checkpoints
WHERE exchange = ? AND symbol = ? AND start_time = ?
""", (exchange, symbol, start_time))
row = cursor.fetchone()
conn.close()
if row:
return MigrationCheckpoint(
exchange=row[0],
symbol=row[1],
start_time=row[2],
end_time=row[3],
last_trade_id=row[4],
total_records=row[5],
checksum=row[6],
last_updated=datetime.fromisoformat(row[7])
)
return None
async def fetch_bybit_trades(self, symbol: str,
from_id: Optional[str] = None,
limit: int = 1000) -> List[Dict]:
"""Holt Bybit-Handelsdaten mit optionaler ID-basierter Pagination"""
url = "https://api.bybit.com/v5/market/recent-trade"
params = {
"category": "spot",
"symbol": symbol,
"limit": limit
}
if from_id:
params["from"] = from_id
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 429:
await asyncio.sleep(60) # Rate-Limit Handhabung
return await self.fetch_bybit_trades(symbol, from_id, limit)
data = await response.json()
return data.get("result", {}).get("list", [])
async def fetch_binance_trades(self, symbol: str,
from_id: Optional[int] = None,
limit: int = 1000) -> List[Dict]:
"""Holt Binance-Handelsdaten mit ID-basierter Pagination"""
url = "https://api.binance.com/api/v3/trades"
params = {
"symbol": symbol,
"limit": limit
}
if from_id:
params["fromId"] = from_id
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
data = await response.json()
if isinstance(data, dict) and "code" in data:
raise Exception(f"Binance API Error: {data['msg']}")
return data
async def migrate_exchange_data(self, exchange: str, symbol: str,
start_date: datetime, end_date: datetime):
"""Führt die vollständige Migration mit Resume-Funktionalität durch"""
start_time_ms = int(start_date.timestamp() * 1000)
checkpoint = self.load_checkpoint(exchange, symbol, start_time_ms)
current_id = checkpoint.last_trade_id if checkpoint else None
total_fetched = checkpoint.total_records if checkpoint else 0
print(f"🚀 Starte Migration: {exchange}/{symbol}")
print(f" Letzter Checkpoint: {checkpoint.last_updated if checkpoint else 'Keiner'}")
batch_count = 0
max_batches = 10000
while batch_count < max_batches:
try:
if exchange == "bybit":
trades = await self.fetch_bybit_trades(symbol, current_id)
else:
current_id_int = int(current_id) if current_id else None
trades = await self.fetch_binance_trades(symbol, current_id_int)
if not trades:
print(f"✅ Migration abgeschlossen: {exchange}/{symbol}")
break
# Hier würden Sie die Daten in HolySheep speichern
await self.store_to_holysheep(trades, exchange, symbol)
current_id = str(trades[-1].get("id") or trades[-1].get("tradeId", ""))
total_fetched += len(trades)
batch_count += 1
# Checkpoint alle 100 Batches speichern
if batch_count % 100 == 0:
checkpoint = MigrationCheckpoint(
exchange=exchange,
symbol=symbol,
start_time=start_time_ms,
end_time=int(datetime.now().timestamp() * 1000),
last_trade_id=current_id,
total_records=total_fetched,
checksum=hashlib.md5(str(current_id).encode()).hexdigest(),
last_updated=datetime.now()
)
self.save_checkpoint(checkpoint)
print(f" Fortschritt: {total_fetched} Trades verarbeitet")
await asyncio.sleep(0.1) # Respect rate limits
except Exception as e:
print(f"⚠️ Fehler bei Batch {batch_count}: {e}")
await asyncio.sleep(5)
continue
return total_fetched
async def store_to_holysheep(self, trades: List[Dict],
exchange: str, symbol: str):
"""Speichert die normalisierten Daten über HolySheep API"""
base_url = "https://api.holysheep.ai/v1"
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
normalized_trades = [self._normalize_trade(t, exchange) for t in trades]
payload = {
"exchange": exchange,
"symbol": symbol,
"trades": normalized_trades
}
# HolySheep API für Marktdaten-Storage
async with aiohttp.ClientSession() as session:
async with session.post(
f"{base_url}/marketdata/trades",
headers=headers,
json=payload
) as response:
if response.status != 200:
raise Exception(f"HolySheep API Error: {await response.text()}")
def _normalize_trade(self, trade: Dict, exchange: str) -> Dict:
"""Normalisiert Handelsdaten in ein einheitliches Format"""
if exchange == "bybit":
return {
"id": trade["id"],
"price": float(trade["price"]),
"quantity": float(trade["qty"]),
"quote_quantity": float(trade["quote_qty"]),
"timestamp": int(trade["ts"]),
"is_buyer_maker": trade.get("is_buyer_maker", False),
"is_best_match": trade.get("is_best_match", True)
}
else: # Binance
return {
"id": str(trade["id"]),
"price": float(trade["price"]),
"quantity": float(trade["qty"]),
"quote_quantity": float(trade["quoteQty"]),
"timestamp": trade["time"],
"is_buyer_maker": trade["isBuyerMaker"],
"is_best_match": trade["isBestMatch"]
}
async def main():
migrator = TickDataMigrator("migration_state.db")
# Migration für BTCUSDT auf beiden Börsen
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
for symbol in symbols:
for exchange in ["bybit", "binance"]:
await migrator.migrate_exchange_data(
exchange=exchange,
symbol=symbol,
start_date=datetime(2024, 1, 1),
end_date=datetime.now()
)
if __name__ == "__main__":
asyncio.run(main())2. Deduplizierung (去重策略)
Bei der Kombination von Daten aus verschiedenen Quellen entstehen zwangsläufig Duplikate. Hier ist eine bewährte Deduplizierungsstrategie:
#!/usr/bin/env python3
"""
Trade Data Deduplication Engine
Normalisiert und vereinheitlicht Tick-Daten von multiple Quellen
"""
import pandas as pd
from typing import List, Dict, Tuple
from datetime import datetime
import hashlib
class TradeDeduplicator:
def __init__(self, time_window_ms: int = 100):
"""
Args:
time_window_ms: Zeitfenster in Millisekunden für die
Zusammenführung von Trades
"""
self.time_window = time_window_ms
def generate_trade_key(self, trade: Dict) -> str:
"""Generiert einen eindeutigen Schlüssel basierend auf
Trade-Charakteristika"""
# Für Bybit
if "id" in trade and "ts" in trade:
key_data = f"{trade['id']}_{trade['ts']}"
# Für Binance
elif "tradeId" in trade and "time" in trade:
key_data = f"{trade['tradeId']}_{trade['time']}"
else:
# Fallback: Hash aus Preis, Menge und Zeit
key_data = f"{trade['price']}_{trade['quantity']}_{trade['timestamp']}"
return hashlib.sha256(key_data.encode()).hexdigest()[:16]
def merge_trades(self, trades: List[Dict]) -> List[Dict]:
"""
Führt Trades mit identischen IDs oder innerhalb des
Zeitfensters zusammen
"""
if not trades:
return []
df = pd.DataFrame(trades)
# Primäre Deduplizierung: Exakte ID-Matches
df["trade_key"] = df.apply(
lambda row: self.generate_trade_key(row.to_dict()), axis=1
)
df_deduped = df.drop_duplicates(subset=["trade_key"], keep="first")
# Sekundäre Deduplizierung: Zeitfenster-basierte Zusammenführung
# Trades innerhalb von 100ms mit gleichem Preis werden zusammengeführt
df_deduped = self._merge_close_trades(df_deduped)
return df_deduped.drop(columns=["trade_key"]).to_dict("records")
def _merge_close_trades(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Fasst Trades zusammen, die innerhalb des Zeitfensters liegen
und denselben Preis haben
"""
if "timestamp" not in df.columns and "ts" not in df.columns:
return df
time_col = "timestamp" if "timestamp" in df.columns else "ts"
df = df.sort_values(time_col)
merged_rows = []
current_group = None
for idx, row in df.iterrows():
row_dict = row.to_dict()
if current_group is None:
current_group = {
"rows": [row_dict],
"start_time": row_dict[time_col]
}
else:
time_diff = row_dict[time_col] - current_group["start_time"]
if (time_diff <= self.time_window and
row_dict.get("price") == current_group["rows"][0].get("price")):
current_group["rows"].append(row_dict)
else:
# Zusammenführen der aktuellen Gruppe
merged_rows.append(self._aggregate_group(
current_group["rows"], time_col
))
current_group = {
"rows": [row_dict],
"start_time": row_dict[time_col]
}
# Letzte Gruppe verarbeiten
if current_group:
merged_rows.append(self._aggregate_group(
current_group["rows"], time_col
))
return pd.DataFrame(merged_rows)
def _aggregate_group(self, group: List[Dict], time_col: str) -> Dict:
"""
Aggregiert eine Gruppe von Trades zu einem einzigen Trade
"""
if len(group) == 1:
return group[0]
# Letzten Trade als Basis verwenden
result = group[-1].copy()
# Gesamtmenge summieren
total_qty = sum(t.get("quantity") or t.get("qty", 0) for t in group)
result["quantity"] = total_qty
result["qty"] = total_qty
# Zeit auf Durchschnitt setzen
avg_time = sum(t[time_col] for t in group) // len(group)
result[time_col] = avg_time
# Anzahl der zusammengeführten Trades vermerken
result["merged_count"] = len(group)
return result
class CrossExchangeNormalizer:
"""Normalisiert Handelsdaten für den Vergleich zwischen Börsen"""
EXCHANGE_CONFIGS = {
"bybit": {
"price_field": "price",
"qty_field": "qty",
"time_field": "ts",
"id_field": "id",
"time_multiplier": 1 # Already in ms
},
"binance": {
"price_field": "price",
"qty_field": "qty",
"time_field": "time",
"id_field": "tradeId",
"time_multiplier": 1 # Already in ms
},
"okx": {
"price_field": "px",
"qty_field": "sz",
"time_field": "ts",
"id_field": "tradeId",
"time_multiplier": 1
}
}
def normalize(self, trade: Dict, exchange: str) -> Dict:
"""Normalisiert Trade in ein einheitliches Format"""
config = self.EXCHANGE_CONFIGS.get(exchange.lower())
if not config:
raise ValueError(f"Unknown exchange: {exchange}")
return {
"exchange": exchange.lower(),
"trade_id": str(trade[config["id_field"]]),
"price": float(trade[config["price_field"]]),
"quantity": float(trade[config["qty_field"]]),
"timestamp_ms": int(trade[config["time_field"]]) * config["time_multiplier"],
"datetime": datetime.fromtimestamp(
int(trade[config["time_field"]]) * config["time_multiplier"] / 1000
),
"side": "sell" if trade.get("is_buyer_maker", False) else "buy"
}
def create_unified_dataframe(self, trades_by_exchange: Dict[str, List[Dict]]) -> pd.DataFrame:
"""
Erstellt einen vereinheitlichten DataFrame aus mehreren Börsen
Args:
trades_by_exchange: Dict mit Börsenname als Key und Liste
von Trades als Value
"""
all_normalized = []
for exchange, trades in trades_by_exchange.items():
for trade in trades:
normalized = self.normalize(trade, exchange)
all_normalized.append(normalized)
df = pd.DataFrame(all_normalized)
df = df.sort_values("timestamp_ms")
df = df.reset_index(drop=True)
return df
def analyze_price_discrepancy(df: pd.DataFrame, symbol: str,
max_discrepancy_pct: float = 0.1) -> Dict:
"""
Analysiert Preisabweichungen zwischen同一Symbol auf verschiedenen Börsen
"""
results = {
"symbol": symbol,
"total_records": len(df),
"exchanges": df["exchange"].unique().tolist(),
"anomalies": [],
"avg_price_by_exchange": {}
}
# Durchschnittspreis pro Börse
for exchange in df["exchange"].unique():
avg_price = df[df["exchange"] == exchange]["price"].mean()
results["avg_price_by_exchange"][exchange] = round(avg_price, 8)
# Zeitfenster-Analyse für Preisabweichungen
df["time_bucket"] = (df["timestamp_ms"] // 1000) * 1000 # Round to seconds
grouped = df.groupby(["time_bucket", "exchange"])["price"].mean().unstack()
if len(grouped.columns) > 1:
# Berechne prozentuale Abweichung zwischen Börsen
price_cols = [col for col in grouped.columns if col != "time_bucket"]
for i, col1 in enumerate(price_cols):
for col2 in price_cols[i+1:]:
discrepancy = abs(grouped[col1] - grouped[col2]) / grouped[col1] * 100
large_discrepancies = discrepancy[discrepancy > max_discrepancy_pct]
if len(large_discrepancies) > 0:
results["anomalies"].append({
"exchange_pair": f"{col1}_vs_{col2}",
"anomaly_count": len(large_discrepancies),
"max_discrepancy_pct": round(discrepancy.max(), 4),
"avg_discrepancy_pct": round(discrepancy.mean(), 4)
})
return results
Beispiel-Verwendung
if __name__ == "__main__":
# Beispiel-Daten von Bybit und Binance
bybit_trades = [
{"id": "1001", "price": "50123.50", "qty": "0.015", "ts": 1712345678000},
{"id": "1002", "price": "50124.00", "qty": "0.020", "ts": 1712345680000},
{"id": "1003", "price": "50125.50", "qty": "0.010", "ts": 1712345685000},
]
binance_trades = [
{"tradeId": "5001", "price": "50123.55", "qty": "0.015", "time": 1712345678000},
{"tradeId": "5002", "price": "50124.00", "qty": "0.025", "time": 1712345680000},
{"tradeId": "5003", "price": "50126.00", "qty": "0.008", "time": 1712345686000},
]
# Deduplizierung
deduplicator = TradeDeduplicator(time_window_ms=200)
all_trades = bybit_trades + binance_trades
deduped = deduplicator.merge_trades(all_trades)
print(f"Nach Deduplizierung: {len(deduped)} Trades (von ursprünglich {len(all_trades)})")
# Normalisierung
normalizer = CrossExchangeNormalizer()
unified_df = normalizer.create_unified_dataframe({
"bybit": bybit_trades,
"binance": binance_trades
})
print("\nVereinheitlichte Daten:")
print(unified_df.to_string())
# Analyse der Preisabweichungen
analysis = analyze_price_discrepancy(unified_df, "BTCUSDT")
print(f"\nPreisanalyse: {analysis}")3. Zeitstempel-Normalisierung (时间戳归一)
Jede Börse verwendet leicht unterschiedliche Zeitformate. Die Normalisierung ist entscheidend für korrekte Marktdaten-Analysen:
- Bybit: Millisekunden-Timestamps (z.B.
1712345678000) - Binance: Millisekunden-Timestamps im
time-Feld - OKX: Mikrosekunden für historische Daten, Millisekunden für Echtzeit
- Coinbase: ISO 8601 Strings oder Unix-Timestamps je nach Endpunkt
Vergleich: Tardis.dev vs. HolySheep AI
| Kriterium | Tardis.dev | HolySheep AI | Vorteil |
|---|---|---|---|
| Preis pro 1M Trades | $15.00 - $25.00 | $1.50 - $3.00 | HolySheep (85%+ günstiger) |
| Latenz (P99) | 120-250ms | <50ms | HolySheep |
| Rate-Limits | Streng (100 req/min) | Grosszügig (10.000 req/min) | HolySheep |
| Breakpoint-Resume | Manuell implementieren | Integriert mit automatischer Wiederholung | HolySheep |
| Deduplizierung | SQL-basiert (extern) | API-seitig mit nativer Unterstützung | HolySheep |
| Zahlungsmethoden | Nur Kreditkarte/PayPal | WeChat Pay, Alipay, Kreditkarte | HolySheep |
| Kostenloses Kontingent | 10.000 Trades | 100.000 Trades + $5 Startguthaben | HolySheep |
| Bybit-Support | Spot & Futures | Spot, Futures, Optionen, Perpetuals | HolySheep |
| Binance-Support | Spot & USDⓈ-M | Vollständig (Spot, COIN-M, USDⓈ-M) | HolySheep |
| Historie-Tiefe | 2 Jahre | 5+ Jahre | HolySheep |
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Quantitative Trader: Die Kombination aus niedriger Latenz (<50ms) und kostengünstigen Preisen macht HolySheep ideal für Strategien, die grosse Datenmengen verarbeiten
- HFT-Firmen: Wer Millisekunden-geschwindigkeit benötigt, profitiert von der dedizierten Infrastruktur
- Research-Teams: Akademische Forscher und Data Scientists erhalten grosszügige Freikontingente für Experimente
- Multi-Exchange-Strategien: Die einheitliche API für Bybit, Binance und weitere Börsen vereinfacht die Entwicklung
- Budget-bewusste Startups: Mit ¥1=$1-Preisgestaltung und WeChat/Alipay-Unterstützung ideal für chinesische Teams
❌ Weniger geeignet für:
- Institutionelle Tier-1-Händler: Wer dediziertePrivate-Connectivity und SLA-Garantien über 99.99% benötigt, sollte spezialisierte Bloomberg/Refinitiv-Lösungen in Betracht ziehen
- Regulierte Finanzinstitute: Wenn Compliance-spezifische Audit-Trails und regulatorische Zertifizierungen erforderlich sind
- OTC/Dark-Pool-Analyse: Für Daten ausserhalb der regulären Börsen-Infrastruktur
Preise und ROI
Die Kostenanalyse zeigt ein überzeugendes Bild für HolySheep AI:
| Szenario | Tardis.dev Kosten | HolySheep AI Kosten | Ersparnis |
|---|---|---|---|
| 10M Trades/Monat | $150 - $250 | $15 - $30 | ~90% |
| 100M Trades/Monat | $1.500 - $2.500 | $150 - $300 | ~90% |
| 1B Trades/Monat | $15.000 - $25.000 | $1.500 - $3.000 | ~90% |
ROI-Kalkulation für ein typisches Quant-Team:
- Jährliche Datenkosten (Tardis): $18.000 - $30.000
- Jährliche Datenkosten (HolySheep): $1.800 - $3.600
- Direkte Ersparnis: $16.200 - $26.400/Jahr
- Indirekte Ersparnis (Entwicklungszeit für Resume/Dedup): ~40 Stunden à $150 = $6.000
- Gesamt-Jahresersparnis: $22.200 - $32.400
Warum HolySheep wählen
- Unschlagbare Preisgestaltung: Mit ¥1=$1-Wechselkurs und 85%+ günstigeren Preisen als Konkurrenten ist HolySheep die wirtschaftlichste Lösung für datenintensive Anwendungen
- Asiatische Zahlungsmethoden: WeChat Pay und Alipay-Unterstützung machen es für chinesische Trader und Firmen attraktiv – keine internationalen Kreditkarten erforderlich
- <50ms Latenz: Für zeitkritische Strategien ist die niedrige Latenz ein entscheidender Wettbewerbsvorteil
- Kostenlose Credits: $5 Startguthaben + 100.000 kostenlose Trades ermöglichen umfangreiche Tests ohne Vorabkosten
- Integrierte Datennormalisierung: Breakpoint-Resume, Deduplizierung und Zeitstempel-Normalisierung sind nativ in der API implementiert
- Breite Börsenabdeckung: Bybit, Binance, OKX, Huobi und weitere – eine API für alle wichtigen asiatischen Kryptobörsen
- AI-Integration: Zusätzlich zu Marktdaten können Sie GPT-4.1 ($8/MTok), Claude Sonnet 4.5 ($15/MTok), Gemini 2.5 Flash ($2.50/MTok) und DeepSeek V3.2 ($0.42/MTok) für sentimentale Analyse und prädiktive Modelle nutzen
Migrations-Rollback-Plan
Ein vollständiger Migrationsplan muss immer einen Rollback-Pfad enthalten:
#!/usr/bin/env python3
"""
Migrations-Rollback-Skript für Notfälle
Stellt den vorherigen Zustand bei Problemen wieder her
"""
import sqlite3
import json
from datetime import datetime
from typing import Dict, List
import shutil
import os
class MigrationRollback:
def __init__(self, backup_dir: str = "./migration_backups"):
self.backup_dir = backup_dir
os.makedirs(backup_dir, exist_ok=True)
def create_backup(self, source_db: str, label: str) -> str:
"""
Erstellt ein vollständiges Backup der Quelldatenbank
Args:
source_db: Pfad zur Quelldatenbank
label: Eindeutige Kennung für das Backup
Returns:
Pfad zum Backup-File
"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"{label}_{timestamp}_backup.db"
backup_path = os.path.join(self.backup_dir, backup_name)
shutil.copy2(source_db, backup_path)
# Auch die Checkpoint-Tabelle separat speichern
checkpoint_backup = f"{backup_path}.checkpoints.json"
if os.path.exists("migration_state.db"):
conn = sqlite3.connect("migration_state.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM checkpoints")
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
with open(checkpoint_backup, "w") as f:
json.dump({"columns": columns, "data": rows}, f, indent=2)
conn.close()
print(f"✅ Backup erstellt: {backup_path}")
return backup_path
def restore_backup(self, backup_path: str, target_db: str):
"""
Stellt eine vorherige Sicherung wieder her
Args:
backup_path: Pfad zum Backup
target_db: Zielpfad für die Wiederherstellung
"""
if not os.path.exists(backup_path):
raise FileNotFoundError(f"Backup nicht gefunden: {backup_path}")
shutil.copy2(backup_path, target_db)
# Checkpoints wiederherstellen
checkpoint_file = f"{backup_path}.checkpoints.json"
if os.path.exists(checkpoint_file):
conn = sqlite3.connect("migration_state.db")
cursor = conn.cursor()
with open(checkpoint_file, "r") as f:
checkpoint_data = json
Verwandte Ressourcen
Verwandte Artikel