Dify插件开发：集成交易所实时数据API – Vollständiger Leitfaden 2026

Einleitung: Der Moment, der alles veränderte

Was ist Dify und warum eignet es sich für API-Integrationen?

• Modularität: Jedes Plugin erfüllt eine spezifische Aufgabe – Datenabruf, Analyse, Alert-Versand
• Fehlertoleranz: Integrierte Retry-Mechanismen und Fallback-Strategien
• Skalierbarkeit: Plugins lassen sich wiederverwenden und kombinieren
• Observability: Eingebaute Logging- und Monitoring-Funktionen

mein-trading-plugin/
├── app/
│   └── plugins/
│       └── exchange_connector/
│           ├── __init__.py          # Plugin-Manifest
│           ├── executor.py          # Hauptlogik
│           ├── api_client.py        # HTTP-Client für Exchange
│           ├── schema.py            # Datenmodelle
│           └── config.py            # Konfiguration
├── pyproject.toml
└── README.md

Plugin-Entwicklung Schritt für Schritt

Schritt 1: Projektstruktur und Abhängigkeiten definieren

[project]
name = "exchange-realtime-connector"
version = "1.0.0"
description = "Dify plugin for real-time exchange data"
requires-python = ">=3.10"

dependencies = [
    "httpx>=0.27.0",
    "pydantic>=2.0.0",
    "asyncio-loop-around>=0.17.0",
    "holysheep-ai>=1.0.0",  # Optional: AI-Anreicherung
]

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23.0", "ruff>=0.4.0"]

[tool.dify.plugin]
identifier = "com.holysheep.exchange.realtime"
name = "Exchange Realtime Connector"
version = "1.0.0"
description = "Retrieves real-time trading data from cryptocurrency exchanges"
icon = "assets/icon.png"
category = "data"
entry = "app/plugins/exchange_connector/__init__.py"

Schritt 2: Das Plugin-Manifest erstellen

# app/plugins/exchange_connector/__init__.py
"""
HolySheep AI Exchange Realtime Connector Plugin für Dify
Version: 1.0.0
"""

from typing import TYPE_CHECKING
from dify_app import DifyPlugin, DifyPluginManifest
from dify_app.file import File
from .executor import ExchangeRealtimeExecutor
from .schema import (
    ExchangeInput,
    ExchangeOutput,
    SupportedExchange,
    TimeFrame,
)

if TYPE_CHECKING:
    from dify_plugin import ToolInvokeMetadata

manifest = DifyPluginManifest(
    identifier="com.holysheep.exchange.realtime",
    version="1.0.0",
    name="Exchange Realtime Connector",
    description="Ruft Echtzeit-Kurse und Marktdaten von Kryptowährungsbörsen ab",
    icon="assets/icon.png",
    tags=["trading", "crypto", "realtime", "market-data"],
    api_schema=ExchangeInput,
    output_schema=ExchangeOutput,
)

class ExchangeRealtimePlugin(DifyPlugin):
    def __init__(self):
        super().__init__(manifest)
        self.executor = ExchangeRealtimeExecutor()

    def invoke(self, parameters: dict, **kwargs) -> dict:
        """Haupteinstiegspunkt für Dify-Workflows"""
        return self.executor.execute(parameters)

plugin = ExchangeRealtimePlugin()

Schritt 3: Den API-Client für Exchange-Daten implementieren

# app/plugins/exchange_connector/api_client.py
"""
Exchange API Client mit Retry-Logik und Fallback
"""

import asyncio
import time
from typing import Optional
from dataclasses import dataclass
import httpx
from .schema import (
    TickerData,
    OrderBookData,
    SupportedExchange,
    TimeFrame,
)


@dataclass
class APIResponse:
    """Standardisierte API-Antwort"""
    success: bool
    data: Optional[dict] = None
    error: Optional[str] = None
    latency_ms: float = 0.0
    retry_count: int = 0


class ExchangeAPIClient:
    """Async HTTP-Client mit automatischer Retry-Logik"""
    
    BASE_URLS = {
        SupportedExchange.BINANCE: "https://api.binance.com/api/v3",
        SupportedExchange.COINBASE: "https://api.coinbase.com/v2",
        SupportedExchange.KUCOIN: "https://api.kucoin.com/api/v1",
    }
    
    TIMEOUT_SECONDS = 10
    MAX_RETRIES = 3
    RETRY_DELAYS = [1, 3, 7]  # Sekunden zwischen Retry-Versuchen
    
    def __init__(self, exchange: SupportedExchange, api_key: str = "", secret: str = ""):
        self.exchange = exchange
        self.api_key = api_key
        self.secret = secret
        self.base_url = self.BASE_URLS.get(exchange, "")
        
        # Timeout-Client mit Connection Pooling
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(self.TIMEOUT_SECONDS),
            limits=httpx.Limits(max_keepalive_connections=10, max_connections=20),
            follow_redirects=True,
        )
    
    async def get_ticker(self, symbol: str) -> APIResponse:
        """
        Ruft Echtzeit-Kursdaten für ein Trading-Paar ab
        
        Args:
            symbol: z.B. 'BTC/USDT' oder 'BTCUSDT'
        
        Returns:
            APIResponse mit Ticker-Daten oder Fehlerinformationen
        """
        endpoint = self._get_endpoint("ticker", symbol)
        start_time = time.perf_counter()
        
        for attempt in range(self.MAX_RETRIES):
            try:
                headers = self._build_headers()
                response = await self._client.get(endpoint, headers=headers)
                
                if response.status_code == 200:
                    data = self._parse_ticker_response(response.json())
                    latency = (time.perf_counter() - start_time) * 1000
                    return APIResponse(
                        success=True,
                        data=data,
                        latency_ms=latency,
                        retry_count=attempt
                    )
                    
                elif response.status_code == 429:
                    # Rate Limit – länger warten
                    await asyncio.sleep(self.RETRY_DELAYS[attempt] * 2)
                    continue
                    
                elif response.status_code == 401:
                    return APIResponse(
                        success=False,
                        error=f"401 Unauthorized: API-Key ungültig oder abgelaufen",
                        retry_count=attempt
                    )
                    
                else:
                    return APIResponse(
                        success=False,
                        error=f"HTTP {response.status_code}: {response.text[:200]}",
                        retry_count=attempt
                    )
                    
            except httpx.TimeoutException as e:
                if attempt == self.MAX_RETRIES - 1:
                    return APIResponse(
                        success=False,
                        error=f"ConnectionError: timeout after {self.TIMEOUT_SECONDS}s",
                        retry_count=attempt
                    )
                await asyncio.sleep(self.RETRY_DELAYS[attempt])
                
            except httpx.ConnectError as e:
                if attempt == self.MAX_RETRIES - 1:
                    return APIResponse(
                        success=False,
                        error=f"ConnectionError: failed to connect – {str(e)}",
                        retry_count=attempt
                    )
                await asyncio.sleep(self.RETRY_DELAYS[attempt])
                
            except Exception as e:
                return APIResponse(
                    success=False,
                    error=f"Unexpected error: {type(e).__name__}: {str(e)}",
                    retry_count=attempt
                )
        
        return APIResponse(
            success=False,
            error="Max retries exceeded",
            retry_count=self.MAX_RETRIES
        )
    
    async def get_order_book(self, symbol: str, limit: int = 20) -> APIResponse:
        """Ruft Orderbook-Daten ab (Bid/Ask-Preise)"""
        endpoint = self._get_endpoint("orderbook", symbol, limit=limit)
        start_time = time.perf_counter()
        
        try:
            response = await self._client.get(endpoint, timeout=self.TIMEOUT_SECONDS)
            
            if response.status_code == 200:
                data = self._parse_orderbook_response(response.json())
                latency = (time.perf_counter() - start_time) * 1000
                return APIResponse(
                    success=True,
                    data=data,
                    latency_ms=latency
                )
            else:
                return APIResponse(
                    success=False,
                    error=f"HTTP {response.status_code}",
                    retry_count=0
                )
                
        except Exception as e:
            return APIResponse(
                success=False,
                error=str(e),
                retry_count=0
            )
    
    def _get_endpoint(self, data_type: str, symbol: str, **params) -> str:
        """Generiert Exchange-spezifische Endpoints"""
        symbol = self._normalize_symbol(symbol)
        
        if self.exchange == SupportedExchange.BINANCE:
            if data_type == "ticker":
                return f"{self.base_url}/ticker/24hr?symbol={symbol}"
            elif data_type == "orderbook":
                return f"{self.base_url}/depth?symbol={symbol}&limit={params.get('limit', 20)}"
                
        elif self.exchange == SupportedExchange.COINBASE:
            if data_type == "ticker":
                return f"{self.base_url}/prices/{symbol}/spot"
            elif data_type == "orderbook":
                return f"{self.base_url}/book/{symbol}"
                
        elif self.exchange == SupportedExchange.KUCOIN:
            if data_type == "ticker":
                return f"{self.base_url}/market/orderbook/level1?symbol={symbol}"
            elif data_type == "orderbook":
                return f"{self.base_url}/market/orderbook/level2?symbol={symbol}"
        
        return f"{self.base_url}/{data_type}"
    
    def _normalize_symbol(self, symbol: str) -> str:
        """Normalisiert Symbol-Format je nach Exchange"""
        symbol = symbol.upper().replace("/", "")
        
        if self.exchange == SupportedExchange.BINANCE:
            return symbol
        elif self.exchange == SupportedExchange.COINBASE:
            return f"{symbol[:3]}-{symbol[3:]}"
        elif self.exchange == SupportedExchange.KUCOIN:
            return symbol
        return symbol
    
    def _build_headers(self) -> dict:
        """Baut Authentifizierungs-Header"""
        headers = {
            "Content-Type": "application/json",
            "User-Agent": "HolySheep-Dify-Plugin/1.0",
            "X-Market-Data-Client": "dify-plugin",
        }
        
        if self.api_key:
            headers["X-API-KEY"] = self.api_key
            
        return headers
    
    def _parse_ticker_response(self, data: dict) -> dict:
        """Parst Exchange-spezifische Antworten in Standardformat"""
        if self.exchange == SupportedExchange.BINANCE:
            return {
                "symbol": data.get("symbol", ""),
                "price": float(data.get("lastPrice", 0)),
                "price_change_24h": float(data.get("priceChange", 0)),
                "price_change_percent_24h": float(data.get("priceChangePercent", 0)),
                "high_24h": float(data.get("highPrice", 0)),
                "low_24h": float(data.get("lowPrice", 0)),
                "volume_24h": float(data.get("volume", 0)),
                "quote_volume_24h": float(data.get("quoteVolume", 0)),
                "timestamp": data.get("closeTime", 0),
            }
        # Weitere Exchange-spezifische Parser...
        return data
    
    def _parse_orderbook_response(self, data: dict) -> dict:
        """Parst Orderbook-Daten"""
        return {
            "bids": data.get("bids", data.get("data", {}).get("bids", []))[:10],
            "asks": data.get("asks", data.get("data", {}).get("asks", []))[:10],
            "timestamp": data.get("timestamp", 0),
        }
    
    async def close(self):
        """Räumt Ressourcen auf"""
        await self._client.aclose()
    
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

Schritt 4: Den Executor für Workflow-Integration erstellen

# app/plugins/exchange_connector/executor.py
"""
Workflow-Executor für Dify-Integration
"""

import asyncio
from typing import Optional
from .api_client import ExchangeAPIClient
from .schema import (
    ExchangeInput,
    ExchangeOutput,
    SupportedExchange,
    DataType,
    TickerData,
    OrderBookData,
)


class ExchangeRealtimeExecutor:
    """Hauptlogik für Dify-Workflow-Integration"""
    
    # HolySheep AI Integration für KI-gestützte Marktanalyse
    HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self):
        self._clients = {}  # Cache für API-Clients
    
    def _get_client(self, exchange: SupportedExchange, api_key: str = "", secret: str = "") -> ExchangeAPIClient:
        """Holt oder erstellt einen gecachten API-Client"""
        cache_key = f"{exchange.value}:{api_key[:8] if api_key else 'anonymous'}"
        
        if cache_key not in self._clients:
            self._clients[cache_key] = ExchangeAPIClient(
                exchange=exchange,
                api_key=api_key,
                secret=secret
            )
        
        return self._clients[cache_key]
    
    async def execute(self, parameters: dict) -> dict:
        """
        Haupteinstiegspunkt für Dify-Workflows
        
        Args:
            parameters: Dictionary mit Exchange-Input-Daten
            
        Returns:
            Dictionary mit ExchangeOutput-Daten
        """
        try:
            # Validierung der Eingabeparameter
            input_data = ExchangeInput(**parameters)
        except Exception as e:
            return {
                "success": False,
                "error": f"Invalid parameters: {str(e)}",
                "data": None
            }
        
        # API-Client holen
        client = self._get_client(
            exchange=input_data.exchange,
            api_key=input_data.api_key or "",
            secret=input_data.secret or ""
        )
        
        try:
            # Daten basierend auf angefordertem Typ abrufen
            if input_data.data_type == DataType.TICKER:
                result = await self._fetch_ticker(client, input_data.symbol)
                
            elif input_data.data_type == DataType.ORDERBOOK:
                result = await self._fetch_orderbook(
                    client, 
                    input_data.symbol, 
                    input_data.limit or 20
                )
                
            elif input_data.data_type == DataType.MULTIPLE:
                result = await self._fetch_multiple(client, input_data.symbol)
                
            else:
                result = {"success": False, "error": "Unsupported data type"}
            
            # Optional: KI-Anreicherung via HolySheep AI
            if result.get("success") and input_data.ai_analysis:
                result = await self._enhance_with_ai(result, input_data)
            
            return result
            
        except Exception as e:
            return {
                "success": False,
                "error": f"Execution failed: {type(e).__name__}: {str(e)}",
                "data": None
            }
        finally:
            # Clients nicht schließen (Connection Pooling)
            pass
    
    async def _fetch_ticker(self, client: ExchangeAPIClient, symbol: str) -> dict:
        """Ruft Ticker-Daten ab"""
        response = await client.get_ticker(symbol)
        
        if response.success:
            return {
                "success": True,
                "data_type": "ticker",
                "data": response.data,
                "latency_ms": round(response.latency_ms, 2),
                "retry_count": response.retry_count,
                "exchange_timestamp": response.data.get("timestamp", 0),
            }
        else:
            return {
                "success": False,
                "error": response.error,
                "retry_count": response.retry_count,
            }
    
    async def _fetch_orderbook(self, client: ExchangeAPIClient, symbol: str, limit: int) -> dict:
        """Ruft Orderbook-Daten ab"""
        response = await client.get_order_book(symbol, limit)
        
        if response.success:
            return {
                "success": True,
                "data_type": "orderbook",
                "data": response.data,
                "latency_ms": round(response.latency_ms, 2),
            }
        else:
            return {
                "success": False,
                "error": response.error,
            }
    
    async def _fetch_multiple(self, client: ExchangeAPIClient, symbol: str) -> dict:
        """Ruft mehrere Datentypen parallel ab"""
        ticker_task = client.get_ticker(symbol)
        ob_task = client.get_order_book(symbol, 10)
        
        ticker_result, ob_result = await asyncio.gather(
            ticker_task, ob_task, return_exceptions=True
        )
        
        return {
            "success": True,
            "data_type": "multiple",
            "ticker": ticker_result.data if isinstance(ticker_result, type(response := object())) and hasattr(ticker_result, 'data') else None,
            "orderbook": ob_result.data if isinstance(ob_result, type(response)) and hasattr(ob_result, 'data') else None,
        }
    
    async def _enhance_with_ai(self, data: dict, input_data: ExchangeInput) -> dict:
        """
        Optionale KI-Anreicherung der Marktdaten
        
        Nutzt HolySheep AI für Sentiment-Analyse und Trading-Signale
        """
        try:
            import httpx
            
            # Daten für KI-Analyse aufbereiten
            prompt = self._build_analysis_prompt(data, input_data)
            
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f"{self.HOLYSHEEP_BASE_URL}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {input_data.holysheep_api_key or 'YOUR_HOLYSHEEP_API_KEY'}",
                        "Content-Type": "application/json",
                    },
                    json={
                        "model": "gpt-4.1",
                        "messages": [
                            {"role": "system", "content": "Du bist ein erfahrener Krypto-Marktanalyst."},
                            {"role": "user", "content": prompt}
                        ],
                        "temperature": 0.3,
                        "max_tokens": 500,
                    },
                    timeout=30.0,
                )
                
                if response.status_code == 200:
                    result = response.json()
                    data["ai_analysis"] = {
                        "summary": result["choices"][0]["message"]["content"],
                        "model": result.get("model", "unknown"),
                        "usage": result.get("usage", {}),
                    }
                else:
                    data["ai_warning"] = f"KI-Analyse fehlgeschlagen: {response.status_code}"
                    
        except Exception as e:
            data["ai_warning"] = f"KI-Integration nicht verfügbar: {str(e)}"
        
        return data
    
    def _build_analysis_prompt(self, data: dict, input_data: ExchangeInput) -> str:
        """Baut den Prompt für die KI-Analyse"""
        symbol = input_data.symbol
        
        if input_data.data_type == DataType.TICKER and "data" in data:
            ticker = data["data"]
            prompt = f"""Analysiere die folgenden Marktdaten für {symbol}:

Preis: ${ticker.get('price', 'N/A')}
24h Änderung: {ticker.get('price_change_percent_24h', 'N/A')}%
24h Hoch: ${ticker.get('high_24h', 'N/A')}
24h Tief: ${ticker.get('low_24h', 'N/A')}
Volumen: {ticker.get('volume_24h', 'N/A')}

Gib eine kurze technische Analyse mit:
1. Trendrichtung (bullish/bearish/neutral)
2. Key Support/Resistance Levels
3. Kurzfristige Handelssignale
"""
        else:
            prompt = f"Analysiere die Marktdaten für {symbol} und gib Handelsempfehlungen."
        
        return prompt

Schritt 5: Schema-Definitionen

# app/plugins/exchange_connector/schema.py
"""
Pydantic-Schemata für Eingabe/Ausgabe-Validierung
"""

from enum import Enum
from typing import Optional, List, Literal
from pydantic import BaseModel, Field, field_validator


class SupportedExchange(str, Enum):
    """Unterstützte Kryptowährungsbörsen"""
    BINANCE = "binance"
    COINBASE = "coinbase"
    KUCOIN = "kucoin"
    # Weitere Exchanges einfach hinzufügen


class DataType(str, Enum):
    """Verfügbare Datentypen"""
    TICKER = "ticker"
    ORDERBOOK = "orderbook"
    TRADES = "trades"
    MULTIPLE = "multiple"
    OHLCV = "ohlcv"  # Open/High/Low/Close/Volume


class TimeFrame(str, Enum):
    """Zeitrahmen für historische Daten"""
    MINUTE_1 = "1m"
    MINUTE_5 = "5m"
    MINUTE_15 = "15m"
    HOUR_1 = "1h"
    HOUR_4 = "4h"
    DAY_1 = "1d"
    WEEK_1 = "1w"


class ExchangeInput(BaseModel):
    """Eingabe-Schema für Dify-Workflows"""
    
    exchange: SupportedExchange = Field(
        default=SupportedExchange.BINANCE,
        description="Die Kryptowährungsbörse für Datenabruf"
    )
    
    symbol: str = Field(
        ...,
        description="Trading-Paar, z.B. 'BTC/USDT' oder 'BTCUSDT'",
        examples=["BTC/USDT", "ETH/USDT", "SOL/USDT"],
        min_length=3,
        max_length=20,
    )
    
    data_type: DataType = Field(
        default=DataType.TICKER,
        description="Welche Daten abgerufen werden sollen"
    )
    
    api_key: Optional[str] = Field(
        default=None,
        description="API-Key der Börse (optional für öffentliche Endpoints)",
        validate_default=True,
    )
    
    secret: Optional[str] = Field(
        default=None,
        description="API-Secret der Börse (optional)",
    )
    
    limit: Optional[int] = Field(
        default=20,
        description="Anzahl der Ergebnisse (z.B. Orderbook-Einträge)",
        ge=1,
        le=100,
    )
    
    timeframe: Optional[TimeFrame] = Field(
        default=TimeFrame.HOUR_1,
        description="Zeitrahmen für historische Daten",
    )
    
    ai_analysis: bool = Field(
        default=False,
        description="Ob KI-gestützte Marktanalyse via HolySheep AI aktiviert werden soll",
    )
    
    holysheep_api_key: Optional[str] = Field(
        default=None,
        description="HolySheep AI API-Key für KI-Anreicherung",
    )
    
    @field_validator("symbol")
    @classmethod
    def validate_symbol(cls, v: str) -> str:
        """Validiert und normalisiert Symbol-Format"""
        v = v.upper().strip()
        
        # Akzeptiere BTC/USDT, BTCUSDT, BTC-USDT
        v = v.replace("-", "/")
        
        if "/" not in v and len(v) >= 6:
            # Vermutlich kompaktes Format wie BTCUSDT
            for base in ["USDT", "USDC", "BUSD", "BTC", "ETH"]:
                if v.endswith(base):
                    return f"{v[:-len(base)]}/{base}"
        
        return v
    
    class Config:
        json_schema_extra = {
            "example": {
                "exchange": "binance",
                "symbol": "BTC/USDT",
                "data_type": "ticker",
                "ai_analysis": True,
            }
        }


class ExchangeOutput(BaseModel):
    """Ausgabe-Schema für Dify-Workflows"""
    
    success: bool = Field(..., description="Ob die Anfrage erfolgreich war")
    
    data_type: Optional[str] = Field(None, description="Art der abgerufenen Daten")
    
    data: Optional[dict] = Field(None, description="Die abgerufenen Marktdaten")
    
    latency_ms: Optional[float] = Field(None, description="API-Latenz in Millisekunden")
    
    retry_count: int = Field(default=0, description="Anzahl der Retry-Versuche")
    
    error: Optional[str] = Field(None, description="Fehlermeldung bei Misserfolg")
    
    ai_analysis: Optional[dict] = Field(None, description="KI-gestützte Marktanalyse")
    
    timestamp: int = Field(default_factory=lambda: int(__import__("time").time() * 1000))


class TickerData(BaseModel):
    """Strukturierte Ticker-Daten"""
    
    symbol: str
    price: float
    price_change_24h: float
    price_change_percent_24h: float
    high_24h: float
    low_24h: float
    volume_24h: float
    quote_volume_24h: float
    timestamp: int


class OrderBookData(BaseModel):
    """Strukturierte Orderbook-Daten"""
    
    symbol: str
    bids: List[List[float]] = Field(description="Kauforders [[Preis, Menge], ...]")
    asks: List[List[float]] = Field(description="Verkaufsorders [[Preis, Menge], ...]")
    timestamp: int

Integration in Dify-Workflows

# Beispiel: Dify Workflow JSON-Definition
{
  "nodes": [
    {
      "id": "exchange-input",
      "type": "parameter",
      "data": {
        "variable": "trading_pair",
        "type": "text",
        "default": "BTC/USDT"
      }
    },
    {
      "id": "exchange-connector",
      "type": "tool",
      "data": {
        "provider": "exchange-realtime-connector",
        "method": "invoke",
        "parameters": {
          "exchange": "binance",
          "symbol": "{{trading_pair}}",
          "data_type": "ticker",
          "ai_analysis": true
        }
      }
    },
    {
      "id": "price-check",
      "type": "condition",
      "data": {
        "conditions": [
          {
            "variable": "exchange-connector.success",
            "operator": "equals",
            "value": true
          }
        ]
      }
    }
  ],
  "edges": [
    {
      "source": "exchange-input",
      "target": "exchange-connector"
    },
    {
      "source": "exchange-connector",
      "target": "price-check"
    }
  ]
}

Häufige Fehler und Lösungen

# Timeout-Konfiguration anpassen
class ExchangeAPIClient:
    TIMEOUT_SECONDS = 30  # Erhöht von 10 auf 30
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=10.0,    # Connection-Timeout
                read=30.0,       # Read-Timeout
                write=10.0,      # Write-Timeout
                pool=5.0         # Pool-Timeout
            )
        )

Alternativ: Retry mit exponentieller Backoff
RETRY_DELAYS = [1, 2, 4, 8, 16]  # Sekunden

async def fetch_with_backoff(client, url, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = await client.get(url)
            return response
        except (httpx.TimeoutException, httpx.ConnectError):
            if attempt < max_retries - 1:
                delay = RETRY_DELAYS[attempt]
                await asyncio.sleep(delay)
            else:
                raise

# Authentifizierung validieren und debuggen
async def validate_api_key(exchange: str, api_key: str, secret: str) -> bool:
    """Validiert API-Credentials vor der Verwendung"""
    
    if not api_key or len(api_key) < 10:
        raise ValueError("API-Key zu kurz oder leer")
    
    # Test-Anfrage an den Exchange
    test_endpoints = {
        "binance": "https://api.binance.com/api/v3/account",
        "coinbase": "https://api.coinbase.com/v2/user",
        "kucoin": "https://api.kucoin.com/api/v1/accounts",
    }
    
    headers = {"X-API-KEY": api_key}
    if secret:
        headers["X-API-SECRET"] = secret
        headers["X-API-PASSPHRASE"] = ""  # Manche Exchanges benötigen Passphrase
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                test_endpoints.get(exchange),
                headers=headers,
                timeout=10.0
            )
            
            if response.status_code == 200:
                return True
            elif response.status_code == 401:
                # Versuche, detaillierte Fehlermeldung zu extrahieren
                error_data = response.json()
                error_msg = error_data.get("msg", error_data.get("message", "Unknown"))
                raise ValueError(f"API-Authentifizierung fehlgeschlagen: {error_msg}")
            else:
                raise ValueError(f"Unerwarteter Statuscode: {response.status_code}")
                
        except httpx.ConnectError:
            raise ConnectionError(f"Verbindung zu {exchange} fehlgeschlagen")

# Rate Limiter mit Token Bucket Algorithmus
import asyncio
import time
from collections import deque

class RateLimiter:
    """Token Bucket Rate Limiter für API-Anfragen"""
    
    def __init__(self, requests_per_second: float = 10, burst: int = 20):
        self.rate = requests_per_second
        self.burst = burst
        self.tokens = burst
        self.last_update = time.monotonic()
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        """Wartet bis ein Token verfügbar ist"""
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_update
            
            # Tokens auffüllen basierend auf vergangener Zeit
            self.tokens = min(self.burst, self.tokens + elapsed * self.rate)
            self
Verwandte Ressourcen
📚 KI API Tutorials
💰 Preise ansehen
📖 Entwickler-Dokumentation
🚀 Kostenlos registrieren
Verwandte Artikel
AI 中转站技术架构：负载均衡器选型与配置 — Ein Praxisleitfaden
Node.js 大模型 API SDK 选型指南: Der komplette Vergleich für Entwic
API Rate Limiter Implementierung: Sliding Window vs Token Bu