Deribit作为全球领先的加密货币期权交易所，其历史成交数据与orderbook信息是量化交易回测的核心数据源。然而，API调用稳定性差、网络波动频繁、审计追踪缺失等问题长期困扰着量化开发者。在本文azin zeigen wir Ihnen, wie HolySheep AI diese Herausforderungen mit einem intelligenten Retry- und Audit-System löst – und vergleichen die Kosten mit direkten API-Aufrufen.
2026年API-Preise im Vergleich: HolySheep vs. Offizielle APIs
Bevor wir in die technische Implementierung einsteigen, zunächst ein Blick auf die aktuellen Kosten (Stand: Mai 2026):
| Modell | Offizielle API ($/MTok) | HolySheep ($/MTok) | Ersparnis |
|---|---|---|---|
| GPT-4.1 | $8,00 | $1,20* | 85% |
| Claude Sonnet 4.5 | $15,00 | $2,25* | 85% |
| Gemini 2.5 Flash | $2,50 | $0,38* | 85% |
| DeepSeek V3.2 | $0,42 | $0,06* | 86% |
*Alle Preise basieren auf dem Wechselkurs ¥1 ≈ $1. WeChat und Alipay Zahlung möglich.
Kostenvergleich für 10M Token/Monat
| Szenario | Offizielle API | HolySheep AI |
|---|---|---|
| GPT-4.1 (10M Tok) | $80,00 | $12,00 |
| Claude Sonnet 4.5 (10M Tok) | $150,00 | $22,50 |
| Gemini 2.5 Flash (10M Tok) | $25,00 | $3,80 |
| DeepSeek V3.2 (10M Tok) | $4,20 | $0,60 |
Das Problem: Instabile Deribit-API für historische Daten
Bei der Arbeit mit Deribit-Optionsdaten stoßen wir regelmäßig auf folgende Herausforderungen:
- Rate Limiting: Historische Trades und Orderbook-Deltas sind streng limitiert
- Netzwerkinstabilität: Connection Timeouts bei Peak-Zeiten
- Datenlücken: Fehlende Timestamps in historischen Datensätzen
- Fehlende Audit-Trails: Keine Nachverfolgbarkeit bei fehlgeschlagenen Requests
Die Lösung: HolySheep AI Proxy mit intelligentem Retry-System
HolySheep AI bietet einen stabilen Daten-Proxy mit <50ms Latenz, automatischer Retry-Logik und vollständiger Audit-Funktionalität. Jetzt registrieren und kostenlose Credits sichern!
Architektur-Übersicht
┌─────────────────────────────────────────────────────────────┐
│ HolySheep AI Proxy │
├─────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Retry │───▶│ Audit │───▶│ Rate │ │
│ │ Manager │ │ Logger │ │ Limiter │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Connection Pool Manager │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ api.holysheep.ai/v1 (base_url) │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
Implementierung: Stabiler Retry-Mechanismus
Der folgende Python-Code zeigt eine produktionsreife Implementierung mit exponentiellem Backoff und jitter:
import requests
import time
import json
import hashlib
from datetime import datetime
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
class RetryStrategy(Enum):
EXPONENTIAL_BACKOFF = "exponential"
LINEAR = "linear"
FIBONACCI = "fibonacci"
@dataclass
class RetryConfig:
max_retries: int = 5
base_delay: float = 0.1 # Sekunden
max_delay: float = 30.0
strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF
jitter: bool = True
retry_on_status: tuple = (429, 500, 502, 503, 504)
@dataclass
class AuditEntry:
request_id: str
timestamp: str
endpoint: str
method: str
status_code: Optional[int]
retry_count: int
response_time_ms: float
error: Optional[str] = None
request_hash: str = ""
def to_dict(self) -> Dict[str, Any]:
return {
"request_id": self.request_id,
"timestamp": self.timestamp,
"endpoint": self.endpoint,
"method": self.method,
"status_code": self.status_code,
"retry_count": self.retry_count,
"response_time_ms": round(self.response_time_ms, 2),
"error": self.error,
"request_hash": self.request_hash
}
class HolySheepRetryClient:
"""
Stabiler API-Client mit Retry-Logik und Audit-Trail für Deribit-Daten.
base_url: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str,
retry_config: RetryConfig = None,
audit_log_path: str = "audit_log.jsonl"
):
self.api_key = api_key
self.retry_config = retry_config or RetryConfig()
self.audit_log_path = audit_log_path
self.audit_entries: list[AuditEntry] = []
def _generate_request_id(self) -> str:
"""Eindeutige Request-ID für Audit-Trail"""
timestamp = datetime.utcnow().isoformat()
return hashlib.sha256(f"{timestamp}{self.api_key}".encode()).hexdigest()[:16]
def _calculate_delay(self, attempt: int) -> float:
"""Berechne Delay basierend auf Retry-Strategie"""
base = self.retry_config.base_delay
if self.retry_config.strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
delay = base * (2 ** attempt)
elif self.retry_config.strategy == RetryStrategy.LINEAR:
delay = base * attempt
elif self.retry_config.strategy == RetryStrategy.FIBONACCI:
# Fibonacci-ähnliches Delay
fib = [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
delay = base * fib[min(attempt, 9)]
else:
delay = base
delay = min(delay, self.retry_config.max_delay)
# Jitter hinzufügen für bessere Verteilung
if self.retry_config.jitter:
import random
delay = delay * (0.5 + random.random())
return delay
def _log_audit(self, entry: AuditEntry):
"""Schreibe Audit-Entry in Log-Datei"""
self.audit_entries.append(entry)
try:
with open(self.audit_log_path, "a") as f:
f.write(json.dumps(entry.to_dict()) + "\n")
except IOError as e:
print(f"Audit-Log Fehler: {e}")
def _make_request(
self,
method: str,
endpoint: str,
params: Dict = None,
data: Dict = None,
retry_count: int = 0
) -> tuple[Optional[Dict], AuditEntry]:
"""Einzelner Request mit Timing"""
request_id = self._generate_request_id()
url = f"{self.BASE_URL}/{endpoint.lstrip('/')}"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
start_time = time.perf_counter()
try:
response = requests.request(
method=method,
url=url,
params=params,
json=data,
headers=headers,
timeout=30
)
elapsed_ms = (time.perf_counter() - start_time) * 1000
entry = AuditEntry(
request_id=request_id,
timestamp=datetime.utcnow().isoformat(),
endpoint=endpoint,
method=method,
status_code=response.status_code,
retry_count=retry_count,
response_time_ms=elapsed_ms,
error=None
)
return response.json() if response.ok else None, entry
except requests.exceptions.Timeout as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
entry = AuditEntry(
request_id=request_id,
timestamp=datetime.utcnow().isoformat(),
endpoint=endpoint,
method=method,
status_code=None,
retry_count=retry_count,
response_time_ms=elapsed_ms,
error=f"Timeout: {str(e)}"
)
return None, entry
except requests.exceptions.RequestException as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
entry = AuditEntry(
request_id=request_id,
timestamp=datetime.utcnow().isoformat(),
endpoint=endpoint,
method=method,
status_code=None,
retry_count=retry_count,
response_time_ms=elapsed_ms,
error=f"RequestException: {str(e)}"
)
return None, entry
def request_with_retry(
self,
method: str,
endpoint: str,
params: Dict = None,
data: Dict = None
) -> Optional[Dict]:
"""
Request mit automatischer Retry-Logik und Audit-Trail.
Gibt Response-Dict zurück oder None bei permanentem Fehler.
"""
last_error = None
for attempt in range(self.retry_config.max_retries + 1):
response, entry = self._make_request(
method=method,
endpoint=endpoint,
params=params,
data=data,
retry_count=attempt
)
self._log_audit(entry)
if response is not None:
return response
# Prüfe ob Retry sinnvoll ist
status = entry.status_code
if status and status not in self.retry_config.retry_on_status:
# Kein Retry bei Client-Fehlern (4xx außer 429)
if status < 400 or status == 429:
if attempt < self.retry_config.max_retries:
delay = self._calculate_delay(attempt)
print(f"Retry {attempt+1}/{self.retry_config.max_retries} "
f"nach {delay:.2f}s für {endpoint}")
time.sleep(delay)
continue
last_error = f"Status {status} - kein Retry"
break
if attempt < self.retry_config.max_retries:
delay = self._calculate_delay(attempt)
print(f"Retry {attempt+1}/{self.retry_config.max_retries} "
f"nach {delay:.2f}s - Error: {entry.error}")
time.sleep(delay)
else:
last_error = entry.error
print(f"Permanenter Fehler nach {self.retry_config.max_retries} Retries: {last_error}")
return None
Beispiel-Nutzung
client = HolySheepRetryClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
retry_config=RetryConfig(
max_retries=5,
base_delay=0.1,
max_delay=10.0,
strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
jitter=True
)
)
Deribit历史数据获取：完整示例
Jetzt kombinieren wir den Retry-Client mit Deribit-spezifischen Endpunkten für historische Optionsdaten:
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Optional
class DeribitHistoricalData:
"""
Holistischer Zugriff auf Deribit historische Daten über HolySheep Proxy.
Unterstützt: Trades, Orderbook Snapsots, Index-Preise, Volatility Data
"""
def __init__(self, holy_sheep_client: HolySheepRetryClient):
self.client = holy_sheep_client
def get_trade_history(
self,
instrument_name: str,
start_timestamp: int,
end_timestamp: int,
count: int = 1000
) -> Optional[pd.DataFrame]:
"""
Rufe historische Trades für ein Deribit-Instrument ab.
Args:
instrument_name: z.B. "BTC-28MAR25-95000-C"
start_timestamp: Unix ms
end_timestamp: Unix ms
count: Max Anzahl Trades
Returns:
DataFrame mit Trades oder None
"""
params = {
"instrument_name": instrument_name,
"start_timestamp": start_timestamp,
"end_timestamp": end_timestamp,
"count": count
}
# Retry-Mechanismus automatisch aktiv
response = self.client.request_with_retry(
method="GET",
endpoint="/deribit/historical/trades",
params=params
)
if response and "result" in response:
trades = response["result"]["trades"]
df = pd.DataFrame(trades)
if not df.empty:
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
return df
return None
def get_orderbook_snapshot(
self,
instrument_name: str,
depth: int = 10
) -> Optional[Dict]:
"""
Rufe aktuellen Orderbook-Snapshot ab.
Returns:
Dict mit bids, asks, timestamp
"""
params = {
"instrument_name": instrument_name,
"depth": depth
}
response = self.client.request_with_retry(
method="GET",
endpoint="/deribit/orderbook/snapshot",
params=params
)
if response and "result" in response:
return response["result"]
return None
def get_volatility_index(
self,
currency: str = "BTC",
period: str = "1h"
) -> Optional[pd.DataFrame]:
"""
Rufe historische Volatility-Index-Daten ab (wichtig für Options-Pricing).
"""
params = {
"currency": currency,
"period": period
}
response = self.client.request_with_retry(
method="GET",
endpoint="/deribit/volatility/index",
params=params
)
if response and "result" in response:
data = response["result"]["data"]
return pd.DataFrame(data)
return None
def batch_get_trades(
self,
instruments: List[str],
start_time: datetime,
end_time: datetime
) -> Dict[str, pd.DataFrame]:
"""
Hole Trades für mehrere Instrumente parallel.
Returns: Dict mit instrument_name -> DataFrame
"""
results = {}
start_ms = int(start_time.timestamp() * 1000)
end_ms = int(end_time.timestamp() * 1000)
for instrument in instruments:
df = self.get_trade_history(
instrument_name=instrument,
start_timestamp=start_ms,
end_timestamp=end_ms
)
if df is not None:
results[instrument] = df
else:
print(f"Warnung: Keine Daten für {instrument}")
return results
=== Beispiel: Backtest für Bitcoin-Option-Strategie ===
def run_backtest_example():
"""Vollständiger Backtest-Beispiel mit Retry und Audit"""
client = HolySheepRetryClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
deribit = DeribitHistoricalData(client)
# Backtest-Periode: Letzte 30 Tage
end_time = datetime.now()
start_time = end_time - timedelta(days=30)
# hole Daten für 3 verschiedene Strikes
instruments = [
"BTC-28MAR25-90000-C", # OTM Call
"BTC-28MAR25-95000-C", # ATM Call
"BTC-28MAR25-100000-C" # Deep OTM Call
]
print("Starte Backtest-Datensammlung...")
all_trades = deribit.batch_get_trades(
instruments=instruments,
start_time=start_time,
end_time=end_time
)
# Zeige Statistiken
for instrument, df in all_trades.items():
print(f"\n{instrument}:")
print(f" Trades: {len(df)}")
print(f" Volumen: {df['quantity'].sum() if 'quantity' in df.columns else 'N/A'}")
print(f" Zeitraum: {df['timestamp'].min()} bis {df['timestamp'].max()}")
# Hole Orderbook-Daten für aktuelle Analyse
ob = deribit.get_orderbook_snapshot("BTC-28MAR25-95000-C")
if ob:
print(f"\nAktueller Orderbook:")
print(f" Bids: {len(ob.get('bids', []))}")
print(f" Asks: {len(ob.get('asks', []))}")
# Hole Volatility-Index für Greeks-Berechnung
vol_df = deribit.get_volatility_index("BTC", "1h")
if vol_df is not None:
print(f"\nVolatility-Daten: {len(vol_df)} Einträge")
print(vol_df.describe())
return all_trades
if __name__ == "__main__":
run_backtest_example()
Audit-System: Vollständige Nachverfolgbarkeit
import pandas as pd
from datetime import datetime, timedelta
class AuditAnalyzer:
"""
Analysiert den Audit-Trail für Performance-Optimierung.
"""
@staticmethod
def load_audit_log(path: str = "audit_log.jsonl") -> pd.DataFrame:
"""Lade alle Audit-Einträge"""
entries = []
with open(path, "r") as f:
for line in f:
entries.append(json.loads(line))
return pd.DataFrame(entries)
@staticmethod
def analyze_retry_patterns(df: pd.DataFrame) -> Dict:
"""Analysiere Retry-Häufigkeit und -Ursachen"""
total_requests = len(df)
requests_with_retry = len(df[df["retry_count"] > 0])
retry_rate = requests_with_retry / total_requests * 100
# Fehler gruppiert nach Endpoint
errors = df[df["error"].notna()]
error_by_endpoint = errors.groupby("endpoint").size()
# Durchschnittliche Response-Zeit
avg_response_time = df["response_time_ms"].mean()
return {
"total_requests": total_requests,
"requests_with_retry": requests_with_retry,
"retry_rate_percent": round(retry_rate, 2),
"error_by_endpoint": error_by_endpoint.to_dict(),
"avg_response_time_ms": round(avg_response_time, 2),
"p95_response_time_ms": round(df["response_time_ms"].quantile(0.95), 2)
}
@staticmethod
def generate_cost_report(
df: pd.DataFrame,
token_cost_per_1k: float = 0.0012
) -> Dict:
"""Schätze API-Kosten basierend auf Request-Zählung"""
# Geschätzte Token pro Request (basierend auf Deribit-Daten)
avg_tokens_per_request = 2500
total_requests = len(df)
estimated_tokens = total_requests * avg_tokens_per_request
# Kosten mit HolySheep (~85% Ersparnis)
holy_sheep_cost = (estimated_tokens / 1_000_000) * token_cost_per_1k
official_cost = holy_sheep_cost / 0.15 # 85% Ersparnis
return {
"total_requests": total_requests,
"estimated_tokens": estimated_tokens,
"holy_sheep_cost_usd": round(holy_sheep_cost, 4),
"official_api_cost_usd": round(official_cost, 4),
"savings_usd": round(official_cost - holy_sheep_cost, 4),
"savings_percent": 85
}
@staticmethod
def export_report(df: pd.DataFrame, output_path: str):
"""Exportiere vollständigen Audit-Report"""
retry_analysis = AuditAnalyzer.analyze_retry_patterns(df)
cost_report = AuditAnalyzer.generate_cost_report(df)
report = {
"generated_at": datetime.now().isoformat(),
"retry_analysis": retry_analysis,
"cost_report": cost_report,
"requests": df.to_dict(orient="records")
}
with open(output_path, "w") as f:
json.dump(report, f, indent=2, default=str)
return report
=== Audit-Analyse nach Backtest ===
def analyze_backtest_audit():
"""Analysiere die Audit-Logs nach einem Backtest"""
df = AuditAnalyzer.load_audit_log("audit_log.jsonl")
# Filter nur Deribit-Requests
deribit_df = df[df["endpoint"].str.contains("deribit", case=False)]
print("=== Deribit API Audit-Analyse ===")
print(f"Zeitraum: {df['timestamp'].min()} bis {df['timestamp'].max()}")
print(f"Gesamt-Requests: {len(deribit_df)}")
# Retry-Analyse
retry_stats = AuditAnalyzer.analyze_retry_patterns(deribit_df)
print(f"\nRetry-Statistik:")
print(f" Retry-Rate: {retry_stats['retry_rate_percent']}%")
print(f" Durchschn. Response-Zeit: {retry_stats['avg_response_time_ms']}ms")
print(f" P95 Response-Zeit: {retry_stats['p95_response_time_ms']}ms")
# Kosten-Analyse
cost_report = AuditAnalyzer.generate_cost_report(deribit_df)
print(f"\nKosten-Analyse:")
print(f" HolySheep Kosten: ${cost_report['holy_sheep_cost_usd']}")
print(f" Offizielle API Kosten: ${cost_report['official_api_cost_usd']}")
print(f" Ersparnis: ${cost_report['savings_usd']} ({cost_report['savings_percent']}%)")
# Exportiere Report
report = AuditAnalyzer.export_report(deribit_df, "backtest_audit_report.json")
print(f"\nReport exportiert nach: backtest_audit_report.json")
return report
Geeignet / Nicht geeignet für
| ✅ Ideal für HolySheep | ⚠️ Eingeschränkt / Alternativen prüfen |
|---|---|
| Quant-Fonds mit hohem API-Volumen (>1M Tok/Monat) | Single-Developer-Projekte mit <100K Tok/Monat |
| Backtesting-Pipelines mit häufigen Retry-Bedarf | Latenz-kritische HFT-Strategien (<10ms Anforderung) |
| Teams, die WeChat/Alipay für Zahlungen nutzen | Unternehmen mit ausschließlich USD/Bank-Transfer |
| Entwickler, die <50ms Latenz benötigen | Regulatorische Umgebungen mit besonderen Compliance-Anforderungen |
| Protobuf-basierte Produktionssysteme | Legacy-Systeme, die ausschließlich gRPC benötigen |
Preise und ROI
Der ROI von HolySheep AI ist besonders bei hohem Volumen überzeugend:
| Volumen/Monat | Offizielle API | HolySheep AI | Jährliche Ersparnis | ROI |
|---|---|---|---|---|
| 100K Tokens | $800 (GPT-4.1) | $120 | $8.160 | 7x günstiger |
| 1M Tokens | $8.000 | $1.200 | $81.600 | 7x günstiger |
| 10M Tokens | $80.000 | $12.000 | $816.000 | 7x günstiger |
| 100M Tokens | $800.000 | $120.000 | $8.160.000 | 7x günstiger |
Break-Even: Selbst bei nur 10.000 Tokens/Monat ($1.200/Jahr vs. $80/Jahr) amortisiert sich die Nutzung, wenn man die kostenlosen Credits und WeChat/Alipay-Zahlungsmöglichkeiten berücksichtigt.
Warum HolySheep wählen
- 85%+ Ersparnis gegenüber offiziellen APIs – Preise ab $0,06/MTok für DeepSeek V3.2
- <50ms Latenz für produktive Trading-Pipelines
- Automatischer Retry mit exponentiellem Backoff und Jitter
- Vollständiger Audit-Trail für Compliance und Debugging
- WeChat & Alipay Zahlungsmöglichkeit für chinesische Teams
- Kostenlose Credits für initiale Tests und Prototyping
- Protobuf-Support für bandwidth-effiziente Produktionssysteme
Häufige Fehler und Lösungen
Fehler 1: Rate Limit trotz Retry
Problem: API gibt 429-Status zurück, aber Retry scheint nicht zu funktionieren.
# FEHLERHAFT: Kein Response-Handling bei 429
response = requests.get(url)
if response.status_code != 200:
time.sleep(1)
response = requests.get(url) # Gleiche URL ohne Änderung
✅ RICHTIG: Rate-Limit-Header auswerten und Retry-Delay anpassen
response = requests.get(url, headers={"X-RateLimit-Retry-After": "0"})
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate Limit erreicht. Warte {retry_after}s...")
time.sleep(retry_after)
# Exponential Backoff mit Rate-Limit-Info
for i in range(max_retries):
response = requests.get(url)
if response.status_code == 200:
break
wait_time = int(response.headers.get("X-RateLimit-Reset", 60))
time.sleep(min(wait_time, 300)) # Max 5 Minuten warten
Fehler 2: Connection Pool Erschöpfung
Problem: "Connection pool exhausted" Fehler bei vielen parallelen Requests.
# FEHLERHAFT: Neue Session für jeden Request
for endpoint in endpoints:
session = requests.Session() # Jedes Mal neue Connection
response = session.get(endpoint)
✅ RICHTIG: Singleton Session mit Connection Pool
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class HolySheepSession:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
adapter = HTTPAdapter(
pool_connections=10, # Anzahl gepoolter Verbindungen
pool_maxsize=100, # Max Pool-Größe
max_retries=Retry(
total=3,
backoff_factor=0.1,
status_forcelist=[429, 500, 502, 503, 504]
)
)
cls._instance.session = requests.Session()
cls._instance.session.mount("https://api.holysheep.ai", adapter)
cls._instance.session.mount("http://", adapter)
return cls._instance
def get_session(self) -> requests.Session:
return self.session
Nutzung:
session_manager = HolySheepSession()
for endpoint in endpoints:
response = session_manager.get_session().get(endpoint)
Fehler 3: Audit-Log Datenverlust bei Crash
Problem: Bei Prozessabsturz gehen alle unbgeschriebenen Audit-Einträge verloren.
# FEHLERHAFT: Keine Buffer-Logik
def log_request(self, entry):
with open("audit.jsonl", "a") as f: # Jeder Write ist teuer
f.write(json.dumps(entry) + "\n")
✅ RICHTIG: Buffered Write mit periodischem Flush
import threading
from queue import Queue
from atexit import register
class BufferedAuditLogger:
def __init__(self, filepath: str, buffer_size: int = 100, flush_interval: int = 5):
self.filepath = filepath
self.buffer_size = buffer_size
self.flush_interval = flush_interval
self.buffer: Queue = Queue()
self.lock = threading.Lock()
self._shutdown = False
# Periodic Flush Thread
self.flush_thread = threading.Thread(target=self._periodic_flush)
self.flush_thread.daemon = True
self.flush_thread.start()
# Finaler Flush bei Prozessende
register(self._emergency_flush)
def _periodic_flush(self):
import time
while not self._shutdown:
time.sleep(self.flush_interval)
self._flush_buffer()
def _flush_buffer(self):
entries = []
while not self.buffer.empty():
try:
entries.append(self.buffer.get_nowait())
except:
break
if entries:
with self.lock:
with open(self.filepath,