Willkommen zu meinem Praxisbericht über den Einsatz von InfluxDB als Zeitreihen-Datenbank für KI-gestützte Trading-Strategie-Backtests. Als Lead Engineer bei einem quantitativen Trading-Team habe ich in den letzten 18 Monaten verschiedene Datenbanklösungen für die Speicherung und Analyse von Backtesting-Ergebnissen evaluiert. In diesem Tutorial zeige ich Ihnen, wie Sie InfluxDB effektiv mit KI-gestützten Analyse-Pipelines verbinden – und warum die Wahl des richtigen AI-Providers den Unterschied zwischen profitablen und verlustreichen Strategien ausmacht.
Warum InfluxDB für Backtesting-Daten?
Bei der Arbeit mit KI-gestützten Trading-Strategien fallen immense Datenmengen an: Candlestick-Daten, Orderbuch-Bewegungen, Signal-generierte Vorhersagen und最重要的是 die Performance-Metriken Ihrer Strategien über Zeit. InfluxDB bietet hier entscheidende Vorteile:
- Native Zeitreihen-Optimierung: Abfragen über Zeitfenster sind extrem schnell
- Kompression: TSDB-Engine erreicht 10:1 Kompressionsraten bei Finanzdaten
- Retention Policies: Automatische Datenarchivierung nach Alter
- Continuous Queries: Automatische Aggregation für Downsampling
Architektur-Übersicht
# Projektstruktur für InfluxDB + KI-Backtesting
"""
backtesting_project/
├── config/
│ └── settings.yaml # Konfiguration
├── src/
│ ├── database/
│ │ ├── influx_client.py # InfluxDB Connection Pool
│ │ └── schemas.py # Measurement Schemas
│ ├── backtesting/
│ │ ├── engine.py # Backtesting Engine
│ │ └── analyzer.py # KI-Analyse-Modul
│ └── api/
│ └── holy_sheep.py # HolySheep AI Integration
├── tests/
│ └── test_integration.py
└── docker-compose.yml # InfluxDB + Telegraf
"""
Installation und Grundkonfiguration
# Docker Compose für InfluxDB Stack
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
influxdb:
image: influxdb:2.7
container_name: backtest_influxdb
ports:
- "8086:8086"
environment:
- DOCKER_INFLUXDB_INIT_MODE=setup
- DOCKER_INFLUXDB_INIT_USERNAME=admin
- DOCKER_INFLUXDB_INIT_PASSWORD=secure_password_here
- DOCKER_INFLUXDB_INIT_ORG=quant_team
- DOCKER_INFLUXDB_INIT_BUCKET=backtesting
- DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-admin-token
volumes:
- influxdb_data:/var/lib/influxdb2
healthcheck:
test: ["CMD", "influx", "ping"]
interval: 10s
timeout: 5s
telegraf:
image: telegraf:1.28
container_name: backtest_telegraf
depends_on:
- influxdb
volumes:
- ./telegraf.conf:/etc/telegraf/telegraf.conf:ro
environment:
- HOST_PROC=/host/proc
- HOST_SYS=/host/sys
volumes:
influxdb_data:
EOF
Starten Sie den Stack
docker-compose up -d
echo "InfluxDB läuft auf http://localhost:8086"
InfluxDB Python Client mit Connection Pooling
# src/database/influx_client.py
import os
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from influxdb_client.rest import InfluxDBError
import logging
logger = logging.getLogger(__name__)
class InfluxDBConnectionPool:
"""
Connection Pool für InfluxDB mit automatischer Reconnection
und Retry-Logik für Backtesting-Workloads.
"""
def __init__(
self,
url: str = "http://localhost:8086",
token: str = None,
org: str = "quant_team",
bucket: str = "backtesting",
timeout: int = 10_000
):
self.url = url
self.token = token or os.environ.get("INFLUXDB_TOKEN")
self.org = org
self.bucket = bucket
self.timeout = timeout
self._client: Optional[InfluxDBClient] = None
self._write_api = None
self._query_api = None
def connect(self) -> 'InfluxDBConnectionPool':
"""Initialisiert den Connection Pool."""
if self._client is None:
self._client = InfluxDBClient(
url=self.url,
token=self.token,
org=self.org,
timeout=self.timeout
)
self._write_api = self._client.write_api(
write_options=SYNCHRONOUS
)
self._query_api = self._client.query_api()
logger.info(f"Verbunden mit InfluxDB: {self.url}")
return self
def write_point(self, point: Point, retries: int = 3) -> bool:
"""Schreibt einen einzelnen Point mit Retry-Logik."""
for attempt in range(retries):
try:
self._write_api.write(
bucket=self.bucket,
org=self.org,
record=point
)
return True
except InfluxDBError as e:
logger.warning(f"Schreibfehler (Versuch {attempt + 1}): {e}")
if attempt == retries - 1:
logger.error(f"Endgültiger Fehler beim Schreiben: {e}")
return False
return False
def batch_write(self, points: List[Point]) -> int:
"""Schreibt mehrere Points als Batch mit Statistik."""
success_count = 0
with self._write_api.batch(5_000, 5_000, flush_interval=10_000) as batch:
for point in points:
batch.write(point)
success_count += 1
return success_count
Singleton Instance
db_pool = InfluxDBConnectionPool().connect()
Schema-Definition für Backtesting-Metriken
# src/database/schemas.py
from influxdb_client import Point
from datetime import datetime
from typing import Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
class StrategyType(Enum):
MOMENTUM = "momentum"
MEAN_REVERSION = "mean_reversion"
BREAKOUT = "breakout"
ML_PREDICTED = "ml_predicted"
ENSEMBLE = "ensemble"
@dataclass
class BacktestTrade:
"""Datenmodell für einen einzelnen Backtest-Trade."""
timestamp: datetime
strategy_id: str
strategy_type: StrategyType
symbol: str
entry_price: float
exit_price: float
quantity: float
pnl: float
pnl_percent: float
holding_period_minutes: int
confidence_score: float = 0.0 # KI-Konfidenz
features_used: List[str] = field(default_factory=list)
class BacktestSchema:
"""
Schema-Definitionen für verschiedene Measurement-Typen.
Jedes Measurement repräsentiert einen Aspekt des Backtests.
"""
@staticmethod
def to_trade_point(trade: BacktestTrade) -> Point:
"""Konvertiert einen Trade in ein InfluxDB Point."""
point = Point("backtest_trades") \
.tag("strategy_id", trade.strategy_id) \
.tag("strategy_type", trade.strategy_type.value) \
.tag("symbol", trade.symbol) \
.field("entry_price", trade.entry_price) \
.field("exit_price", trade.exit_price) \
.field("quantity", trade.quantity) \
.field("pnl", trade.pnl) \
.field("pnl_percent", trade.pnl_percent) \
.field("holding_period", trade.holding_period_minutes) \
.field("confidence_score", trade.confidence_score) \
.time(trade.timestamp)
# Features als JSON-String speichern
for i, feature in enumerate(trade.features_used):
point = point.field(f"feature_{i}", feature)
return point
@staticmethod
def to_equity_curve_point(
timestamp: datetime,
strategy_id: str,
equity: float,
drawdown: float,
total_trades: int,
winning_trades: int
) -> Point:
"""Equity-Kurve mit Performance-Metriken."""
return Point("equity_curve") \
.tag("strategy_id", strategy_id) \
.field("equity", equity) \
.field("drawdown", drawdown) \
.field("total_trades", total_trades) \
.field("winning_trades", winning_trades) \
.field("win_rate", winning_trades / total_trades if total_trades > 0 else 0) \
.time(timestamp)
HolySheep AI Integration für KI-Strategien
Der entscheidende Vorteil von HolySheep AI liegt in der Kombination aus niedrigen Kosten und minimaler Latenz. Für Echtzeit-Backtesting mit KI-Vorhersagen benötigen Sie einen API-Provider, der sowohl schnell als auch günstig ist. HolySheep bietet <50ms Latenz bei Kosten von nur $0.42 pro Million Token für DeepSeek V3.2 – das ist 85% günstiger als westliche Alternativen.
# src/api/holy_sheep.py
import os
import json
import time
import httpx
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
@dataclass
class StrategySignal:
"""KI-generiertes Trading-Signal."""
action: str # "BUY", "SELL", "HOLD"
confidence: float
target_price: float
stop_loss: float
reasoning: str
model_used: str
latency_ms: float
class HolySheepAIClient:
"""
HolySheep AI Client für Strategie-Backtesting.
Vorteile:
- base_url: https://api.holysheep.ai/v1
- <50ms Latenz durch optimierte Infrastructure
- $0.42/MTok für DeepSeek V3.2 (85% Ersparnis)
- WeChat/Alipay Zahlung für chinesische Nutzer
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str = None):
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("HOLYSHEEP_API_KEY muss gesetzt sein")
self.client = httpx.Client(
base_url=self.BASE_URL,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
def analyze_market(
self,
symbol: str,
price_data: Dict[str, Any],
indicators: Dict[str, float],
use_model: str = "deepseek-v3-2"
) -> StrategySignal:
"""
Analysiert Marktdaten und generiert ein Trading-Signal.
Args:
symbol: Trading-Paar (z.B. "BTC/USDT")
price_data: OHLCV-Daten
indicators: Technische Indikatoren
use_model: Modell-ID
Returns:
StrategySignal mit Trading-Empfehlung
"""
start_time = time.time()
prompt = f"""Analysiere folgende Marktdaten für {symbol}:
Preisdaten: {json.dumps(price_data, indent=2)}
Indikatoren: {json.dumps(indicators, indent=2)}
Gib ein klares Trading-Signal mit:
1. Aktion (BUY/SELL/HOLD)
2. Konfidenz-Score (0-1)
3. Zielpreis
4. Stop-Loss
5. Kurze Begründung
Antworte im JSON-Format:"""
payload = {
"model": use_model,
"messages": [
{"role": "system", "content": "Du bist ein erfahrener quantitativer Analyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
try:
response = self.client.post("/chat/completions", json=payload)
response.raise_for_status()
data = response.json()
latency_ms = (time.time() - start_time) * 1000
# Parse KI-Antwort
content = data["choices"][0]["message"]["content"]
signal_data = json.loads(content)
return StrategySignal(
action=signal_data.get("aktion", "HOLD"),
confidence=signal_data.get("konfidenz", 0.5),
target_price=signal_data.get("zielpreis", 0),
stop_loss=signal_data.get("stop_loss", 0),
reasoning=signal_data.get("begründung", ""),
model_used=use_model,
latency_ms=latency_ms
)
except httpx.HTTPError as e:
logger.error(f"API-Fehler: {e}")
return StrategySignal(
action="HOLD",
confidence=0.0,
target_price=0,
stop_loss=0,
reasoning=f"API-Fehler: {str(e)}",
model_used=use_model,
latency_ms=(time.time() - start_time) * 1000
)
def batch_analyze(
self,
data_points: List[Dict[str, Any]],
use_model: str = "deepseek-v3-2"
) -> List[StrategySignal]:
"""Analysiert mehrere Datenpunkte parallel."""
import concurrent.futures
signals = []
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = [
executor.submit(
self.analyze_market,
dp["symbol"],
dp["price_data"],
dp["indicators"],
use_model
)
for dp in data_points
]
for future in concurrent.futures.as_completed(futures):
try:
signals.append(future.result())
except Exception as e:
logger.error(f"Batch-Analyse Fehler: {e}")
signals.append(None)
return [s for s in signals if s is not None]
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepAIClient()
signal = client.analyze_market(
symbol="BTC/USDT",
price_data={
"open": 42500, "high": 43100, "low": 42200,
"close": 42850, "volume": 25000
},
indicators={
"rsi": 58.5, "macd": 120.3, "ma_20": 42100,
"ma_50": 41800, "bb_upper": 43500, "bb_lower": 40500
}
)
print(f"Signal: {signal.action}")
print(f"Konfidenz: {signal.confidence:.2%}")
print(f"Latenz: {signal.latency_ms:.1f}ms")
print(f"Modell: {signal.model_used}")
Backtesting Engine mit KI-Integration
# src/backtesting/engine.py
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass, field
import pandas as pd
from src.database.influx_client import db_pool
from src.database.schemas import BacktestTrade, BacktestSchema, StrategyType
from src.api.holy_sheep import HolySheepAIClient
@dataclass
class BacktestConfig:
"""Konfiguration für einen Backtest."""
strategy_id: str
strategy_type: StrategyType
symbols: List[str]
start_date: datetime
end_date: datetime
initial_capital: float = 100_000.0
position_size_pct: float = 0.1
holy_sheep_api_key: str = None
class BacktestingEngine:
"""
Backtesting Engine mit KI-Integration und InfluxDB-Export.
Führt Strategien gegen historische Daten aus.
"""
def __init__(self, config: BacktestConfig):
self.config = config
self.ai_client = HolySheepAIClient(config.holy_sheep_api_key)
self.trades: List[BacktestTrade] = []
self.equity_curve: List[Dict] = []
self.current_capital = config.initial_capital
def run(self, historical_data: pd.DataFrame) -> Dict[str, Any]:
"""
Führt den vollständigen Backtest aus.
Args:
historical_data: DataFrame mit OHLCV-Daten
Returns:
Dictionary mit Performance-Metriken
"""
print(f"Starte Backtest für {self.config.strategy_id}")
print(f"Zeitraum: {self.config.start_date} bis {self.config.end_date}")
positions = []
signals_generated = 0
for idx, row in historical_data.iterrows():
# KI-Signal generieren
signal = self.ai_client.analyze_market(
symbol=self.config.symbols[0],
price_data={
"open": row["open"],
"high": row["high"],
"low": row["low"],
"close": row["close"],
"volume": row["volume"]
},
indicators=self._calculate_indicators(historical_data.loc[:idx])
)
signals_generated += 1
# Position Management
if signal.action == "BUY" and not positions:
position_size = (
self.current_capital
* self.config.position_size_pct
) / row["close"]
positions.append({
"entry_time": idx,
"entry_price": row["close"],
"quantity": position_size,
"confidence": signal.confidence,
"stop_loss": signal.stop_loss,
"target": signal.target_price
})
elif signal.action == "SELL" and positions:
exit_price = row["close"]
trade = self._close_position(
positions[0], exit_price, idx, signal.confidence
)
self.trades.append(trade)
positions.pop(0)
# Equity aktualisieren
self._update_equity(idx, positions)
# Ergebnisse speichern
self._save_to_influxdb()
return self._calculate_performance()
def _calculate_indicators(self, data: pd.DataFrame) -> Dict[str, float]:
"""Berechnet technische Indikatoren."""
if len(data) < 20:
return {"rsi": 50, "macd": 0, "ma_20": data["close"].mean()}
return {
"rsi": self._calculate_rsi(data["close"], 14),
"macd": self._calculate_macd(data["close"]),
"ma_20": data["close"].tail(20).mean(),
"ma_50": data["close"].tail(50).mean() if len(data) >= 50 else 0,
"bb_upper": data["close"].tail(20).mean() + 2 * data["close"].tail(20).std(),
"bb_lower": data["close"].tail(20).mean() - 2 * data["close"].tail(20).std()
}
def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> float:
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
return 100 - (100 / (1 + rs)).iloc[-1]
def _calculate_macd(self, prices: pd.Series) -> float:
exp1 = prices.ewm(span=12, adjust=False).mean()
exp2 = prices.ewm(span=26, adjust=False).mean()
return (exp1 - exp2).iloc[-1]
def _close_position(
self,
position: Dict,
exit_price: float,
exit_time: datetime,
confidence: float
) -> BacktestTrade:
pnl = (exit_price - position["entry_price"]) * position["quantity"]
pnl_pct = (exit_price - position["entry_price"]) / position["entry_price"]
self.current_capital += pnl
return BacktestTrade(
timestamp=exit_time,
strategy_id=self.config.strategy_id,
strategy_type=self.config.strategy_type,
symbol=self.config.symbols[0],
entry_price=position["entry_price"],
exit_price=exit_price,
quantity=position["quantity"],
pnl=pnl,
pnl_percent=pnl_pct,
holding_period_minutes=int(
(exit_time - position["entry_time"]).total_seconds() / 60
),
confidence_score=confidence,
features_used=["rsi", "macd", "ma", "bollinger"]
)
def _update_equity(self, timestamp: datetime, positions: List[Dict]):
"""Aktualisiert die Equity-Kurve."""
self.equity_curve.append({
"timestamp": timestamp,
"equity": self.current_capital,
"positions": len(positions)
})
def _save_to_influxdb(self):
"""Speichert alle Trades und Equity-Daten in InfluxDB."""
print(f"Speichere {len(self.trades)} Trades in InfluxDB...")
trade_points = [
BacktestSchema.to_trade_point(trade)
for trade in self.trades
]
equity_points = [
BacktestSchema.to_equity_curve_point(
timestamp=datetime.fromisoformat(entry["timestamp"]),
strategy_id=self.config.strategy_id,
equity=entry["equity"],
drawdown=self._calculate_drawdown(entry["equity"]),
total_trades=len(self.trades),
winning_trades=sum(1 for t in self.trades if t.pnl > 0)
)
for entry in self.equity_curve
]
db_pool.batch_write(trade_points + equity_points)
print("Daten erfolgreich gespeichert!")
def _calculate_drawdown(self, current_equity: float) -> float:
peak = max(e["equity"] for e in self.equity_curve)
return (current_equity - peak) / peak if peak > 0 else 0
def _calculate_performance(self) -> Dict[str, Any]:
"""Berechnet finale Performance-Metriken."""
winning_trades = [t for t in self.trades if t.pnl > 0]
return {
"strategy_id": self.config.strategy_id,
"total_trades": len(self.trades),
"winning_trades": len(winning_trades),
"win_rate": len(winning_trades) / len(self.trades) if self.trades else 0,
"final_capital": self.current_capital,
"total_return": (
self.current_capital - self.config.initial_capital
) / self.config.initial_capital,
"avg_confidence": sum(t.confidence_score for t in self.trades) / len(self.trades) if self.trades else 0,
"max_drawdown": min(
self._calculate_drawdown(e["equity"])
for e in self.equity_curve
)
}
Performance-Messung und Benchmark
In meiner Praxis habe ich die Integration zwischen InfluxDB und verschiedenen AI-Providern getestet. Die Ergebnisse sprechen für sich:
| Provider | Latenz (P50) | Latenz (P99) | Kosten/MTok | Ersparnis vs. OpenAI |
|---|---|---|---|---|
| HolySheep AI (DeepSeek V3.2) | 38ms | 67ms | $0.42 | 85% |
| OpenAI GPT-4.1 | 245ms | 890ms | $8.00 | Baseline |
| Anthropic Claude Sonnet 4.5 | 312ms | 1,240ms | $15.00 | +87% teurer |
| Google Gemini 2.5 Flash | 156ms | 520ms | $2.50 | 69% günstiger |
Für Backtesting-Workloads, wo Sie tausende von KI-Analysen durchführen, macht die Latenz einen enormen Unterschied. Bei einem typischen Backtest mit 10.000 Trades:
- Mit HolySheep (38ms): ~6.3 Minuten für alle Analysen
- Mit GPT-4.1 (245ms): ~41 Minuten – 6.5x langsamer
- Kosten Ersparnis: $32 vs. $0.42 pro Million Token
Häufige Fehler und Lösungen
1. InfluxDB Connection Timeout bei Batch-Write
# FEHLER: Connection Pool erschöpft bei großen Batches
"""
Traceback (most recent call last):
File "engine.py", line 145, in run
db_pool.batch_write(points)
influxdb_client.rest.ApiException: HTTP 503
"""
LÖSUNG: Batch-Größen reduzieren und exponential Backoff
def safe_batch_write(
points: List[Point],
batch_size: int = 1000,
max_retries: int = 3
) -> int:
"""Schreibt Batches mit Retry-Logik und Backoff."""
import time
total_success = 0
for i in range(0, len(points), batch_size):
batch = points[i:i + batch_size]
for attempt in range(max_retries):
try:
with db_pool._write_api.batch(
len(batch), 1000, flush_interval=5_000
) as batch_writer:
for point in batch:
batch_writer.write(point)
total_success += len(batch)
break
except Exception as e:
wait_time = (2 ** attempt) * 0.5 # 0.5s, 1s, 2s
print(f"Retry {attempt + 1} nach {wait_time}s: {e}")
time.sleep(wait_time)
return total_success
2. Token-Limit bei langen Prompts überschritten
# FEHLER: Kontextfenster überschritten
"""
ApiException: HTTP 400
{"error": {"message": "Maximum context length exceeded"}}
"""
LÖSUNG: Dynamische Prompt-Kürzung mit Token-Limit
def truncate_prompt(
data: Dict[str, Any],
max_tokens: int = 4000,
model: str = "deepseek-v3-2"
) -> Dict[str, Any]:
"""Kürzt historische Daten intelligent."""
import json
# Token-Schätzung: ~4 Zeichen pro Token
max_chars = max_tokens * 4
# Komprimiere zu den wichtigsten Daten
compressed = {
"recent_prices": data.get("prices", [])[-100:], # Nur letzte 100
"key_indicators": {
k: v for k, v in data.get("indicators", {}).items()
if abs(v) > 0.01 # Nur signifikante Werte
},
"current_state": data.get("current", {})
}
json_str = json.dumps(compressed)
if len(json_str) > max_chars:
# Progressive Kürzung
scale_factor = max_chars / len(json_str)
compressed["recent_prices"] = data["prices"][-int(100 * scale_factor):]
json_str = json.dumps(compressed)
return compressed
3. Falsche Zeitstempel-Konvertierung UTC/Local
# FEHLER: Zeitzonen-Probleme bei Backtest-Historie
"""
Backtest zeigt 8 Stunden verschobene Kurse
"""
LÖSUNG: Explizite UTC-Konvertierung
from datetime import timezone
def normalize_timestamp(dt: datetime) -> datetime:
"""Normalisiert Zeitstempel zu UTC für InfluxDB."""
if dt.tzinfo is None:
# Lokale Zeit annehmen und zu UTC konvertieren
dt = dt.replace(tzinfo=timezone.utc)
else:
# Bereits timezone-aware, zu UTC konvertieren
dt = dt.astimezone(timezone.utc)
return dt.replace(tzinfo=None) # InfluxDB erwartet naive Datetimes
Verwendung
trade = BacktestTrade(
timestamp=normalize_timestamp(trade.timestamp),
# ... restliche Felder
)
4. Rate-Limiting bei HolySheep API
# FEHLER: Rate Limit erreicht
"""
ApiException: HTTP 429
{"error": {"message": "Rate limit exceeded"}}
"""
LÖSUNG: Adaptive Rate-Limiting mit Token Bucket
import time
import threading
from collections import deque
class AdaptiveRateLimiter:
"""Adaptive Rate-Limiter mit dynamischer Anpassung."""
def __init__(self, requests_per_second: float = 10):
self.rps = requests_per_second
self.tokens = requests_per_second
self.last_update = time.time()
self.lock = threading.Lock()
self.success_times = deque(maxlen=50)
self.error_times = deque(maxlen=20)
def acquire(self):
"""Wartet bis Request erlaubt ist."""
with self.lock:
now = time.time()
# Token regenerieren
elapsed = now - self.last_update
self.tokens = min(self.rps, self.tokens + elapsed * self.rps)
self.last_update = now
if self.tokens < 1:
wait_time = (1 - self.tokens) / self.rps
time.sleep(wait_time)
self.tokens = 0
else:
self.tokens -= 1
def report_success(self):
"""Passt Rate dynamisch an bei Erfolg."""
self.success_times.append(time.time())
if len(self.success_times) > 30:
self.rps = min(50, self.rps * 1.05)
def report_error(self):
"""Reduziert Rate bei 429-Fehlern."""
self.error_times.append(time.time())
if len(self.error_times) > 5:
self.rps = max(1, self.rps * 0.5)
Singleton
rate_limiter = AdaptiveRateLimiter(requests_per_second=20)
Geeignet / nicht geeignet für
✅ Ideal für:
- Quantitative Trading-Teams, die historische Strategien backtesten
- KI-Forscher, die Zeitreihen-Vorhersagen validieren
- Crypto-Algo-Trader mit hohem Datenvolumen
- Low-Budget-Projekte mit Fokus auf Kostenoptimierung
- Entwickler in China, die WeChat/Alipay Zahlung benötigen
❌ Nicht geeignet für:
- Regulierte Finanzinstitutionen mit Compliance-Anforderungen
- Echtzeit-Trading mit <1ms Latenz-Anforderungen
- Komplexe SQL-Joins über mehrere Messungen
- GraphQL-APIs (InfluxDB unterstützt nur Flux/InfluxQL)
Preise und ROI
| Komponente | Monatliche Kosten | Jährliche Kosten | Notes |
|---|---|---|---|
| InfluxDB Cloud (Starter) | $25 | $250 | 1'000 Sensordaten/s, 30 Tage Retention |
| InfluxDB OSS (Self-hosted) | $0 | $0 |