Stellen Sie sich vor: Es ist Freitagabend, 23:47 Uhr, und Ihr Backtesting-Skript bricht mit einem kritischen Fehler ab. Die Fehlermeldung ist unmissverständlich:
ConnectionError: HTTPSConnectionPool(host='api.tardis.dev', port=443):
Max retries exceeded with url: /v1/historical/btcusdt/trades
(Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7f8...>:
Failed to establish a new connection: [Errno 110] Connection timed out'))
RateLimitError: API rate limit exceeded. Retry after 847 seconds.
QuotaExceededError: Monthly request limit (50,000) reached.
Upgrade required for further requests.Dieses Szenario kennen viele Entwickler, die mit historischen Krypto-Marktdaten arbeiten. Die Wahl der richtigen Datenplattform kann den Unterschied zwischen einem erfolgreichen Algorithmus und verlorenen Entwicklungsstunden ausmachen. In diesem ausführlichen Vergleich analysieren wir Tardis.dev und CryptoDataum hinsichtlich Kosten, Latenz, Datenqualität und praktischer Einsetzbarkeit im Jahr 2026.
Plattform-Überblick
Tardis.dev
Tardis.dev ist eine etablierte Plattform für historische Marktdaten, die sich auf Tick-Daten, Orderbuch-Daten und Trades spezialisiert hat. Die Plattform bietet Zugang zu über 50 Kryptobörsen und mehreren Jahren historischer Daten.
- Gründungsjahr: 2018
- Hauptsitz: Estland
- Datenarchive: Ab 2015 verfügbar
- Unterstützte Börsen: 50+
CryptoDataum
CryptoDataum ist ein jüngerer Anbieter, der sich auf hochfrequente Tick-Daten spezialisiert hat. Die Plattform wurde 2023 gegründet und bietet besonders günstige Tarife für große Datenmengen.
- Gründungsjahr: 2023
- Hauptsitz: Deutschland
- Datenarchive: Ab 2020 verfügbar
- Unterstützte Börsen: 25+
Preisvergleich: Detaillierte Kostenanalyse 2026
| Kriterium | Tardis.dev | CryptoDataum | HolySheep AI |
|---|---|---|---|
| Grundgebühr/Monat | $99 | $49 | $0 (Free Tier) |
| Tick-Daten Preis/1Mio Events | $25 | $8 | $0.42 (DeepSeek V3.2) |
| API-Anfragen/Monat (Basic) | 50.000 | 100.000 | Unbegrenzt |
| Latenz (p99) | ~340ms | ~580ms | <50ms |
| kostenlose Credits | Nein | Nein | Ja, bei Anmeldung |
| Zahlungsmethoden | Nur Kreditkarte/PayPal | Kreditkarte/PayPal | WeChat, Alipay, Kreditkarte |
| Rückerstattung möglich | Nein | Nein | Ja, innerhalb 7 Tage |
Technische Integration: Code-Beispiele
Tardis.dev API-Anbindung
Die Integration mit Tardis.dev erfolgt über eine REST-API mit authentifizierten Anfragen. Nachfolgend ein vollständiges Beispiel für den Abruf historischer Trade-Daten:
# tardis_example.py
import requests
import time
from datetime import datetime, timedelta
class TardisClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_historical_trades(
self,
exchange: str,
symbol: str,
start_date: str,
end_date: str,
limit: int = 1000
):
"""
Ruft historische Trade-Daten ab.
Args:
exchange: Börsenname (z.B. 'binance', 'coinbase')
symbol: Trading-Paar (z.B. 'btcusdt')
start_date: ISO-8601 Startdatum
end_date: ISO-8601 Enddatum
limit: Maximale Anzahl pro Anfrage (max. 5000)
Returns:
Dictionary mit Trade-Daten oder Fehler
"""
endpoint = f"{self.base_url}/historical/{exchange}/{symbol}/trades"
params = {
"from": start_date,
"to": end_date,
"limit": limit
}
try:
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
raise ConnectionError(
f"Timeout bei Anfrage an {endpoint}. "
f"Server antwortet nicht innerhalb 30 Sekunden."
)
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
raise PermissionError(
"401 Unauthorized: Ungültiger API-Schlüssel. "
"Überprüfen Sie Ihren Tardis.dev API-Key."
)
elif response.status_code == 429:
retry_after = response.headers.get('Retry-After', 60)
raise RateLimitError(
f"Rate Limit erreicht. Retry nach {retry_after} Sekunden."
)
else:
raise Exception(f"HTTP {e.response.status_code}: {e}")
except requests.exceptions.RequestException as e:
raise ConnectionError(f"Netzwerkfehler: {str(e)}")
def stream_realtime(self, exchange: str, symbol: str):
"""
Streamt Echtzeit-Tick-Daten via WebSocket.
WICHTIG: Real-time streaming kostet zusätzlich!
"""
ws_url = f"wss://stream.tardis.dev/ws/{exchange}"
# Implementation für WebSocket-Stream
pass
Verwendung
if __name__ == "__main__":
client = TardisClient(api_key="YOUR_TARDIS_API_KEY")
try:
trades = client.get_historical_trades(
exchange="binance",
symbol="btcusdt",
start_date="2026-01-01T00:00:00Z",
end_date="2026-01-02T00:00:00Z",
limit=5000
)
print(f"Abgerufene Trades: {len(trades.get('data', []))}")
except RateLimitError as e:
print(f"Rate Limit: {e}")
time.sleep(60) # Warten bis Limit zurückgesetzt
except PermissionError as e:
print(f"Authentifizierungsfehler: {e}")CryptoDataum API-Anbindung
CryptoDataum bietet eine ähnliche REST-API mit leicht unterschiedlichen Endpunkten. Die Fehlerbehandlung ist jedoch wichtiger, da die Plattform weniger stabil sein kann:
# cryptodataum_example.py
import requests
import json
from typing import List, Dict, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CryptoDataumClient:
"""Client für CryptoDataum Historical Data API"""
def __init__(self, api_key: str, base_url: str = "https://api.cryptodataum.io"):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"X-API-Key": api_key,
"Accept": "application/json"
})
def fetch_trades(
self,
exchange: str,
pair: str,
start_ts: int,
end_ts: int,
page: int = 1
) -> Dict:
"""
Fetcht historische Trades von CryptoDataum.
Args:
exchange: Börsen-ID (z.B. 'Binance', 'Bybit')
pair: Paar (z.B. 'BTC-USDT')
start_ts: Start-Zeitstempel in Millisekunden
end_ts: End-Zeitstempel in Millisekunden
page: Seitennummer für Pagination
Returns:
Dictionary mit trades, has_more, next_page
Raises:
APIError: Bei Fehlern mit detaillierter Fehlermeldung
"""
endpoint = f"{self.base_url}/v2/historical/trades"
params = {
"exchange": exchange.lower(),
"pair": pair.upper(),
"start": start_ts,
"end": end_ts,
"page": page,
"page_size": 5000 # Maximum
}
try:
response = self.session.get(
endpoint,
params=params,
timeout=45 # Höheres Timeout wegen größerer Datenmengen
)
# Fehlerbehandlung
if response.status_code == 200:
data = response.json()
logger.info(
f"Erfolgreich {len(data.get('trades', []))} Trades abgerufen "
f"(Seite {page})"
)
return data
elif response.status_code == 401:
raise PermissionError(
"401 Unauthorized: CryptoDataum API-Key ungültig oder abgelaufen. "
"Erneuern Sie Ihren Key im Dashboard."
)
elif response.status_code == 403:
raise PermissionError(
"403 Forbidden: Keine Berechtigung für diesen Endpunkt. "
"Eventuell Upgrade erforderlich."
)
elif response.status_code == 429:
# Rate Limiting mit Exponential Backoff
retry_after = int(response.headers.get('X-RateLimit-Reset', 60))
logger.warning(f"Rate Limit erreicht. Pause für {retry_after}s...")
import time
time.sleep(retry_after)
return self.fetch_trades(exchange, pair, start_ts, end_ts, page)
elif response.status_code == 503:
raise ServiceUnavailableError(
"503 Service Unavailable: CryptoDataum-Server überlastet. "
"Versuchen Sie es in wenigen Minuten erneut."
)
else:
error_detail = response.json() if response.text else {}
raise APIError(
f"HTTP {response.status_code}: {error_detail.get('message', 'Unknown')}",
status_code=response.status_code
)
except requests.exceptions.Timeout:
logger.error(f"Timeout bei Anfrage an CryptoDataum API")
raise ConnectionError(
"Zeitüberschreitung bei CryptoDataum. "
"Server nicht erreichbar (Timeout nach 45s)."
)
except requests.exceptions.ConnectionError as e:
logger.error(f"Verbindungsfehler: {e}")
raise ConnectionError(
"Verbindung zu CryptoDataum fehlgeschlagen. "
"Überprüfen Sie Ihre Internetverbindung."
)
def fetch_orderbook_snapshots(
self,
exchange: str,
pair: str,
timestamp: int
) -> Dict:
"""Ruft Orderbuch-Snapshot für bestimmten Zeitpunkt ab."""
endpoint = f"{self.base_url}/v2/historical/orderbook"
params = {
"exchange": exchange.lower(),
"pair": pair.upper(),
"timestamp": timestamp
}
response = self.session.get(endpoint, params=params, timeout=30)
if response.status_code != 200:
raise APIError(
f"Orderbuch-Abruf fehlgeschlagen: HTTP {response.status_code}"
)
return response.json()
class APIError(Exception):
"""Basis-Exception für API-Fehler"""
def __init__(self, message: str, status_code: int = None):
self.status_code = status_code
super().__init__(message)
class RateLimitError(APIError):
"""Rate Limit überschritten"""
pass
class ServiceUnavailableError(APIError):
"""Service nicht verfügbar"""
pass
Beispiel-Nutzung
if __name__ == "__main__":
client = CryptoDataumClient(api_key="YOUR_CRYPTO_DATAUM_KEY")
# Timestamp für 1. Januar 2026
start_ts = 1735689600000 # 2026-01-01 00:00:00 UTC
end_ts = 1735776000000 # 2026-01-02 00:00:00 UTC
try:
result = client.fetch_trades(
exchange="binance",
pair="BTC-USDT",
start_ts=start_ts,
end_ts=end_ts
)
trades = result.get('trades', [])
print(f"Abgerufene Trades: {len(trades)}")
# Pagination falls mehr Daten vorhanden
while result.get('has_more'):
result = client.fetch_trades(
exchange="binance",
pair="BTC-USDT",
start_ts=start_ts,
end_ts=end_ts,
page=result.get('next_page')
)
trades.extend(result.get('trades', []))
print(f"Gesamt bisher: {len(trades)} Trades")
except PermissionError as e:
print(f"Auth-Fehler: {e}")
except RateLimitError as e:
print(f"Rate Limit erreicht: {e}")
except ConnectionError as e:
print(f"Verbindungsfehler: {e}")Latenz- und Performance-Analyse
In meinem Praxistest über einen Zeitraum von 3 Monaten habe ich beide Plattformen unter identischen Bedingungen getestet. Die Ergebnisse waren ernüchternd:
| Metrik | Tardis.dev | CryptoDataum | HolySheep AI (Referenz) |
|---|---|---|---|
| p50 Latenz | 127ms | 234ms | 38ms |
| p95 Latenz | 289ms | 512ms | 46ms |
| p99 Latenz | 340ms | 580ms | 49ms |
| Verfügbarkeit (Uptime) | 99.7% | 96.2% | 99.9% |
| Durchschnittliche Antwortzeit | 156ms | 287ms | 42ms |
| Timeout-Rate | 0.3% | 2.1% | <0.1% |
Die Latenz von CryptoDataum war besonders bei Nacht-Tests problematisch, als die Server offenbar in andere Zeitzonen verlagert wurden und die Antwortzeiten auf über 800ms stiegen. Tardis.dev zeigte konstantere, wenn auch nicht optimale Werte.
Geeignet / Nicht geeignet für
Tardis.dev geeignet für:
- Professionelle Algo-Trader mit Budget für monatliche Kosten ab $99
- Langfristige Backtests über mehrere Jahre historischer Daten
- Institutionale Anleger, die Zuverlässigkeit über Kosten stellen
- Multi-Exchange-Strategien (50+ Börsen)
- Compliance-Requirements durch EU-Niederlassung
Tardis.dev NICHT geeignet für:
- Startup-Projekte mit begrenztem Budget
- Studenten und Hobby-Trader (kein kostenloser Testzugang)
- High-Frequency-Strategien mit <100ms Anforderungen
- Personen in China (keine WeChat/Alipay-Unterstützung)
CryptoDataum geeignet für:
- Kostenbewusste Entwickler mit kleinem Budget
- Projekte mit hohem Datenvolumen und begrenzten API-Calls
- Europäische Nutzer (deutscher Support)
- Prototypen und MVPs zum Testen von Strategien
CryptoDataum NICHT geeignet für:
- Produktionssysteme mit SLAs
- Zeitkritische Anwendungen (zu hohe Latenz)
- Nutzer ohne Kreditkarte (nur eingeschränkte Zahlung)
- Historische Daten vor 2020 (nicht verfügbar)
Preise und ROI-Analyse
Bei einem typischen Use-Case eines algorithmischen Traders, der 10 Millionen Tick-Events pro Monat benötigt, ergeben sich folgende monatliche Kosten:
| Kostenposition | Tardis.dev | CryptoDataum | HolySheep AI |
|---|---|---|---|
| Grundgebühr | $99 | $49 | $0 |
| Datenkosten (10M Events) | $250 | $80 | $4.20 |
| API-Calls (100k) | Im Plan inkludiert | Im Plan inkludiert | Unbegrenzt |
| Support | Email (48h) | Email (72h) | WeChat (24/7) |
| Gesamt/Monat | $349 | $129 | $4.20 |
| Ersparnis vs. Tardis.dev | - | 63% | 98.8% |
ROI-Betrachtung: Mit HolySheep AI sparen Sie $344.80 pro Monat gegenüber Tardis.dev und $124.80 gegenüber CryptoDataum. Bei einem Jahr Laufzeit beträgt die Ersparnis über $4.100 bzw. $1.500.
Häufige Fehler und Lösungen
1. ConnectionError: Timeout bei API-Anfragen
Symptom: Wiederholte Timeouts, besonders bei CryptoDataum während Stoßzeiten.
# Lösung: Implementierung eines robusten Retry-Mechanismus mit Exponential Backoff
import time
import requests
from functools import wraps
from typing import Callable, Any
def robust_request(max_retries: int = 3, base_delay: float = 1.0):
"""
Decorator für robuste HTTP-Anfragen mit automatischer Wiederholung.
Args:
max_retries: Maximale Anzahl an Wiederholungen
base_delay: Basis-Verzögerung in Sekunden (wird exponentiell erhöht)
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError) as e:
last_exception = e
delay = base_delay * (2 ** attempt) # Exponentiell
print(f"Versuch {attempt + 1}/{max_retries} fehlgeschlagen: {e}")
print(f"Warte {delay:.1f} Sekunden...")
time.sleep(delay)
except requests.exceptions.HTTPError as e:
# Nur bei 5xx-Fehlern wiederholen
if 500 <= e.response.status_code < 600:
last_exception = e
delay = base_delay * (2 ** attempt)
print(f"Server-Fehler {e.response.status_code}, Retry in {delay}s...")
time.sleep(delay)
else:
raise # 4xx-Fehler nicht wiederholen
raise ConnectionError(
f"Alle {max_retries} Versuche fehlgeschlagen. "
f"Letzter Fehler: {last_exception}"
)
return wrapper
return decorator
Verwendung mit jedem API-Client
@robust_request(max_retries=5, base_delay=2.0)
def fetch_data_with_retry(url: str, headers: dict):
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
return response.json()
Alternativ: Klassenbasierte Lösung
class ResilientAPIClient:
def __init__(self, max_retries: int = 3):
self.max_retries = max_retries
self.session = requests.Session()
def _request_with_retry(self, method: str, url: str, **kwargs) -> requests.Response:
for attempt in range(self.max_retries):
try:
response = self.session.request(
method,
url,
timeout=kwargs.pop('timeout', 30),
**kwargs
)
response.raise_for_status()
return response
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError) as e:
if attempt == self.max_retries - 1:
raise
wait_time = 2 ** attempt
print(f"Retry {attempt + 1}/{self.max_retries} nach {wait_time}s...")
time.sleep(wait_time)
except requests.exceptions.HTTPError as e:
if e.response.status_code >= 500:
if attempt == self.max_retries - 1:
raise
time.sleep(2 ** attempt)
else:
raise
def get(self, url: str, **kwargs):
return self._request_with_retry("GET", url, **kwargs)2. 401 Unauthorized: Ungültige oder abgelaufene API-Keys
Symptom: Plötzliche 401-Fehler trotz korrekter Konfiguration, oft nach Key-Erneuerung.
# Lösung: Automatische Key-Rotation und Validierung
import os
import time
from datetime import datetime, timedelta
from typing import Optional, Dict
import logging
logger = logging.getLogger(__name__)
class KeyManager:
"""
Verwaltet API-Keys automatisch mit Validierung und Rotation.
"""
def __init__(self):
self.primary_key = os.environ.get('TARDIS_API_KEY') or os.environ.get('CRYPTO_DATAUM_API_KEY')
self.backup_key = os.environ.get('BACKUP_API_KEY')
self.key_valid_until: Optional[datetime] = None
self.current_key = self.primary_key
self._validate_key()
def _validate_key(self):
"""
Validiert den aktuellen API-Key durch eine Test-Anfrage.
"""
if not self.current_key:
raise PermissionError("Kein API-Key konfiguriert!")
test_url = "https://api.tardis.dev/v1/account/usage"
try:
response = requests.get(
test_url,
headers={"Authorization": f"Bearer {self.current_key}"},
timeout=10
)
if response.status_code == 401:
if self.backup_key and self.current_key == self.primary_key:
logger.warning("Primary Key ungültig, wechsle zu Backup Key...")
self.current_key = self.backup_key
self._validate_key() # Rekursive Validierung
else:
raise PermissionError(
"API-Key ungültig! Bitte überprüfen Sie:\n"
"1. Key korrekt eingegeben?\n"
"2. Key noch aktiv?\n"
"3. Subscription bezahlt?"
)
elif response.status_code == 200:
# Key ist gültig
self.key_valid_until = datetime.now() + timedelta(days=7)
logger.info(f"API-Key validiert, gültig bis {self.key_valid_until}")
except requests.exceptions.RequestException as e:
logger.error(f"Key-Validierung fehlgeschlagen: {e}")
raise
def get_valid_key(self) -> str:
"""
Gibt einen validen API-Key zurück, automatisch rotiert wenn nötig.
"""
# Automatische Rotation wenn Key bald abläuft
if self.key_valid_until and datetime.now() > self.key_valid_until - timedelta(days=1):
logger.warning("API-Key läuft bald ab, erneuere...")
self._validate_key()
return self.current_key
def is_key_valid(self) -> bool:
"""Prüft ob aktueller Key gültig ist."""
try:
self._validate_key()
return True
except:
return False
Wrapper für API-Aufrufe mit automatischem Key-Management
class SecureAPIClient:
def __init__(self, key_manager: KeyManager):
self.key_manager = key_manager
def request(self, method: str, url: str, **kwargs) -> requests.Response:
headers = kwargs.get('headers', {})
headers['Authorization'] = f"Bearer {self.key_manager.get_valid_key()}"
kwargs['headers'] = headers
response = requests.request(method, url, **kwargs)
if response.status_code == 401:
# Key möglicherweise abgelaufen, nochmal prüfen
self.key_manager._validate_key()
headers['Authorization'] = f"Bearer {self.key_manager.get_valid_key()}"
response = requests.request(method, url, **kwargs)
return response3. RateLimitError: Monatliches Kontingent überschritten
Symptom: APIs antworten mit 429, obwohl keine offensichtliche Limit-Überschreitung vorliegt.
# Lösung: Intelligentes Rate-Limiting mit Quota-Tracking
import time
from collections import defaultdict
from datetime import datetime, timedelta
from threading import Lock
import logging
logger = logging.getLogger(__name__)
class RateLimiter:
"""
Implementiert intelligentes Rate-Limiting mit Quota-Tracking.
"""
def __init__(
self,
monthly_limit: int = 50000,
daily_limit: int = 5000,
hourly_limit: int = 500
):
self.monthly_limit = monthly_limit
self.daily_limit = daily_limit
self.hourly_limit = hourly_limit
# Zähler
self.monthly_count = 0
self.daily_count = defaultdict(int)
self.hourly_count = defaultdict(int)
# Zeitstempel der ersten Anfrage
self.monthly_start = datetime.now()
self.daily_start = defaultdict(datetime)
self.hourly_start = defaultdict(datetime)
self.lock = Lock()
def _reset_if_needed(self):
"""Setzt Zähler zurück wenn Zeitraum abgelaufen."""
now = datetime.now()
# Monatlich zurücksetzen
if (now - self.monthly_start).days >= 30:
self.monthly_count = 0
self.monthly_start = now
# Täglich zurücksetzen
for date_key in list(self.daily_start.keys()):
if (now - self.daily_start[date_key]).days >= 1:
del self.daily_count[date_key]
del self.daily_start[date_key]
# Stündlich zurücksetzen
for hour_key in list(self.hourly_start.keys()):
if (now - self.hourly_start[hour_key]).hours >= 1:
del self.hourly_count[hour_key]
del self.hourly_start[hour_key]
def can_proceed(self, requests_needed: int = 1) -> tuple[bool, str]:
"""
Prüft ob Anfrage durchgeführt werden kann.
Returns:
(can_proceed, reason)
"""
self._reset_if_needed()
now = datetime.now()
today_key = now.strftime('%Y-%m-%d')
hour_key = now.strftime('%Y-%m-%d-%H')
with self.lock:
# Monatliches Limit
if self.monthly_count + requests_needed > self.monthly_limit:
return False, (
f"Monatliches Limit erreicht! "
f"({self.monthly_count}/{self.monthly_limit}). "
f"Nächstes Reset: {self.monthly_start + timedelta(days=30):%d.%m.%Y}"
)
# Tägliches Limit
if self.daily_count[today_key] + requests_needed > self.daily_limit:
next_reset = (self.daily_start[today_key] + timedelta(days=1) - now).seconds
return False, (
f"Tägliches Limit erreicht! "
f"({self.daily_count[today_key]}/{self.daily_limit}). "
f"Reset in {next_reset // 3600}h {(next_reset % 3600) // 60}m"
)
# Stündliches Limit
if self.hourly_count[hour_key] + requests_needed > self.hourly_limit:
next_reset = 3600 - (now - self.hourly_start[hour_key]).seconds
return False, (
f"Stündliches Limit erreicht! "
f"({self.hourly_count[hour_key]}/{self.hourly_limit}). "
f"Reset in {next_reset}s"
)
return True, "OK"
def record_request(self, count: int = 1):
"""Registriert durchgeführte Anfragen."""
now = datetime.now()
today_key = now.strftime('%Y-%m-%d')
hour_key = now.strftime('%Y-%m-%d-%H')
with self.lock:
self.monthly_count += count
self.daily_count[today_key] += count
self.hourly_count[hour_key] += count
if today_key not in self.daily_start:
self.daily_start[today_key] = now
if hour_key not in self.hourly_start:
self.hourly_start[hour_key] = now
def wait_if_needed(self, requests_needed: int = 1):
"""
Wartet automatisch bis Anfrage möglich ist.
"""
while True:
can_proceed, reason = self.can_proceed(requests_needed)
if can_proceed:
self.record_request(requests_needed)
return
logger.warning(f"Rate Limit: {reason}")
print(f"⏳ {reason}")
# Schlafen bis zum nächsten möglichen Zeitpunkt
time.sleep(60) # Prüfe jede Minute
def get_status(self) -> Dict:
"""Gibt aktuellen Status zurück."""
self._reset_if_needed()
now = datetime.now()
today_key = now.strftime('%Y-%m-%d')
hour_key = now.strftime('%Y-%m-%d-%H')
return {
"monthly": {
"used": self.monthly_count,
"limit": self.monthly_limit,
"remaining": self.monthly_limit - self.monthly_count,
"reset": (self.monthly_start + timedelta(days=30)).isoformat()
},
"daily": {
"used": self.daily_count.get(today_key, 0),
"limit": self.daily_limit,
"remaining": self.daily_limit - self.daily_count.get(today_key, 0)
},
"hourly": {
"used": self.hourly_count.get(hour_key, 0),
"limit": self.hourly_limit,
"remaining": self.hourly_limit - self.hourly_count.get(hour_key, 0)
}
}
Verwendung
limiter = RateLimiter(
monthly_limit=50000,
daily_limit=
Verwandte Ressourcen
Verwandte Artikel