Hugging Face's logo

Agueli
/
trading-bot

Model card Files
xet
 Community
1
trading-bot / data /binance.py
aalienau-cmyk
feat: ZIP-Code als neue Basis + vollständiges Linter-Cleanup
e898f62
Raw History Blame Contribute Delete
19.1 kB
"""
binance.py — Kraken Public API Wrapper.
Kein Account nötig, kein API-Key nötig.
Alle Endpoints sind öffentlich.
Inkl. Error Handling und Exponential Backoff bei Rate Limits.
Kraken API (0):
- OHLCV: /0/public/OHLC?pair={symbol}&interval=15
- Ticker: /0/public/Ticker?pair={symbol} (für 24h Volume)
- Pairs: /0/public/AssetPairs (für Liste handelbarer Paare)
Wichtig:
- Kraken nutzt eigene Symbol-Namen (XXBTZUSD statt BTCUSDT)
- Preise/Volumes kommen als Strings, müssen zu float geparst werden
- Volume im Ticker: v[1] = rolling 24h
"""
import time
import logging
import requests
import config
logger = logging.getLogger(__name__)
QUOTE_ASSETS = (
"USDT",
"USDC",
"USD",
"DAI",
"TUSD",
"BUSD",
"GUSD",
"EUR",
"GBP",
"JPY",
"AUD",
"CAD",
"CHF",
)
FIAT_BASE_ASSETS = {"USD", "EUR", "GBP", "JPY", "AUD", "CAD", "CHF", "KRW"}
_ASSET_PAIR_LOOKUP = None
def _asset_to_standard(asset: str) -> str:
"""Normalisiert Kraken-Assetcodes wie XXBT, XETH oder ZEUR."""
asset_map = {
"XXBT": "BTC",
"XBT": "BTC",
"XETH": "ETH",
"XXDG": "DOGE",
"XDG": "DOGE",
"XXRP": "XRP",
"XLTC": "LTC",
"ZUSD": "USD",
"ZEUR": "EUR",
"ZGBP": "GBP",
"ZJPY": "JPY",
"ZAUD": "AUD",
"ZCAD": "CAD",
"ZCHF": "CHF",
"ZKRW": "KRW",
}
if asset in asset_map:
return asset_map[asset]
if len(asset) > 3 and asset[0] in {"X", "Z"}:
return asset[1:]
return asset
def _quote_to_standard(asset: str) -> str:
"""Mappt Kraken-Quotes auf die interne Symbolwelt."""
quote = _asset_to_standard(asset)
if quote == "USD":
return "USDT"
return quote
def _get_asset_pair_lookup() -> dict:
"""Lädt Kraken AssetPairs einmalig für robustes Symbol-Mapping."""
global _ASSET_PAIR_LOOKUP
if _ASSET_PAIR_LOOKUP is None:
url = f"{config.BINANCE_BASE_URL}/0/public/AssetPairs"
data = _request_with_retry(url)
_ASSET_PAIR_LOOKUP = dict(data.get("result", {}))
return _ASSET_PAIR_LOOKUP
class BinanceAPIError(Exception):
"""Fehler bei Kraken API Aufruf."""
pass
class BinanceRateLimitError(BinanceAPIError):
"""Kraken Rate Limit wurde erreicht."""
pass
def _request_with_retry(url: str, params: dict | None = None) -> dict:
"""
GET-Request an Kraken API mit Exponential Backoff.
Raises:
BinanceAPIError: Bei Timeout oder unerwartetem Fehler
BinanceRateLimitError: Bei Rate Limit (429)
"""
backoff = config.BINANCE_BACKOFF_BASE
for attempt in range(config.BINANCE_RETRIES):
try:
response = requests.get(url, params=params, timeout=config.BINANCE_TIMEOUT_SECONDS)
if response.status_code == 429:
logger.warning(
f"Kraken Rate Limit (Versuch {attempt + 1}/{config.BINANCE_RETRIES}). "
f"Warte {backoff:.1f}s..."
)
time.sleep(backoff)
backoff *= 2
continue
response.raise_for_status()
data: dict = response.json()
# Kraken-spezifisch: error-Array mit Inhalt = API-Fehler
errors = data.get("error", [])
# Kraken error ist ein Array von Strings. Wenn nicht Fehler, können
# einfache Warnungen als leere Strings kommen.
real_errors = [e for e in errors if isinstance(e, str) and e.strip()]
if real_errors:
raise BinanceAPIError(f"Kraken API Fehler: {'; '.join(real_errors)}")
return data
except requests.exceptions.Timeout:
logger.warning(
f"Kraken Timeout (Versuch {attempt + 1}/{config.BINANCE_RETRIES}). "
f"Warte {backoff:.1f}s..."
)
if attempt < config.BINANCE_RETRIES - 1:
time.sleep(backoff)
backoff *= 2
else:
raise BinanceAPIError("Kraken API Timeout nach allen Retries") from None
except requests.exceptions.RequestException as e:
raise BinanceAPIError(f"Kraken Request Fehler: {e}") from e
raise BinanceRateLimitError("Kraken Rate Limit — alle Retries aufgebraucht")
def _kraken_to_standard_symbol(kraken_symbol: str) -> str:
"""
Konvertiert Kraken-Symbol zu Standard-Format.
Beispiele:
XXBTZUSD → BTCUSDT
XETHZUSD → ETHUSDT
XXBTZUSD → BTCUSDT
Kraken-Liste (übliche Mappings):
XXBT → BTC, XBT → BTC
ZUSD → USD (Kraken "Z" Präfix für Fiat)
XETH → ETH, XDG → DOGE, XLTC → LTC, XXRP → XRP
Wir konvertieren zu einem Standard-Format für den Rest des Bots.
"""
# Häufigste Kraken-Mappings
prefix_map = {
"XXBT": "BTC",
"XBT": "BTC",
"XETH": "ETH",
"XLTC": "LTC",
"XXRP": "XRP",
"XXDG": "DOGE",
"XADA": "ADA",
"XDOT": "DOT",
"XLINK": "LINK",
"XAVAX": "AVAX",
"XMATIC": "MATIC",
"XSOL": "SOL",
"XUNI": "UNI",
"XAAVE": "AAVE",
"XATOM": "ATOM",
"XETC": "ETC",
"XNEAR": "NEAR",
"XALGO": "ALGO",
"XMANA": "MANA",
"XSAND": "SAND",
"XAXS": "AXS",
"XFTM": "FTM",
"XHBAR": "HBAR",
"XVET": "VET",
"XICP": "ICP",
"XCOMP": "COMP",
"XMKR": "MKR",
"XSNX": "SNX",
"XFI": "FI",
"XRUNE": "RUNE",
"XINJ": "INJ",
"XSEI": "SEI",
"XWIF": "WIF",
"XPEPE": "PEPE",
"XSHIB": "SHIB",
"XBLUR": "BLUR",
"XENJ": "ENJ",
"XMINA": "MINA",
"XGLMR": "GLMR",
"XMOVR": "MOVR",
"XBEAM": "BEAM",
"XCELO": "CELO",
"XONE": "ONE",
"XONT": "ONT",
"XTZ": "XTZ",
"XSTORJ": "STORJ",
"XKSM": "KSM",
"XCHZ": "CHZ",
"XANKR": "ANKR",
"XGRT": "GRT",
"XBAL": "BAL",
"XCRV": "CRV",
"XSUSHI": "SUSHI",
"X1INCH": "1INCH",
"XBNB": "BNB",
"XBC": "XBC",
"XZEC": "ZEC",
"XDASH": "DASH",
"XBSV": "BSV",
"XDOGE": "DOGE",
"XTRX": "TRX",
"XEOS": "EOS",
"XNEO": "NEO",
"XIOTA": "IOTA",
"XWAVES": "WAVES",
"XLSK": "LSK",
"XSTRAT": "STRAT",
"XARK": "ARK",
"XSTEEM": "STEEM",
"XWAX": "WAX",
"XICX": "ICX",
"XIOST": "IOST",
"XREQ": "REQ",
"XPOWER": "POWER",
"XKMD": "KMD",
"XRLC": "RLC",
"XRCN": "RCN",
"XPAY": "PAY",
"XSTORM": "STORM",
"XANT": "ANT",
"XGNO": "GNO",
"XBAT": "BAT",
"XREP": "REP",
"XKNC": "KNC",
"XBNT": "BNT",
"XGNT": "GNT",
"XDAI": "DAI",
"XSAI": "SAI",
"XUSDC": "USDC",
"XUSDT": "USDT",
"XTUSD": "TUSD",
"XPAX": "PAX",
"XGUSD": "GUSD",
"XBUSD": "BUSD",
}
# Kraken-Symbol ist typischerweise 6-8 Zeichen: XXBTZUSD
# Versuche Base und Quote zu extrahieren
# Zuerst: Prüfe ob es ein "Z" oder "X" Präfix hat
# Format: [X/Z][Asset][Z/USD] oder [XBT][ZUSD]
# Spezialfälle zuerst
if kraken_symbol in _KRAKEN_EXCEPTIONS:
return _KRAKEN_EXCEPTIONS[kraken_symbol]
# Versuche das Symbol zu parsen
# Kraken nutzt 3-4 Zeichen für Base und 3-4 für Quote
# ZUSD = USD, ZEUR = EUR, ZGBP = GBP, ZJPY = JPY, ZAUD = AUD, ZCAD = CAD, ZCHF = CHF
# USDT, USDC, DAI, etc. direkt
quote_map = {
"ZUSD": "USDT",
"ZEUR": "EUR",
"ZGBP": "GBP",
"ZJPY": "JPY",
"ZAUD": "AUD",
"ZCAD": "CAD",
"ZCHF": "CHF",
"ZKRW": "KRW",
"USDT": "USDT",
"USDC": "USDC",
"DAI": "DAI",
"TUSD": "TUSD",
"GUSD": "GUSD",
"BUSD": "BUSD",
"PAX": "PAX",
"USD": "USD",
"EUR": "EUR",
}
# Finde das Quote-Asset
quote = None
base = None
for qk, qv in sorted(quote_map.items(), key=lambda x: -len(x[0])):
if kraken_symbol.endswith(qk):
quote = qv
base_part = kraken_symbol[: -len(qk)]
break
if quote is None:
# Fallback: Unbekanntes Quote-Asset
logger.debug(f"Kraken: Unbekanntes Quote-Asset in {kraken_symbol}")
return kraken_symbol
# Base-Asset konvertieren
base = prefix_map.get(base_part, base_part)
return f"{base}{quote}"
def _split_standard_symbol(symbol: str) -> tuple[str, str]:
"""Teilt ein standardisiertes Symbol in Base und Quote."""
for quote in sorted(QUOTE_ASSETS, key=len, reverse=True):
if symbol.endswith(quote):
return symbol[: -len(quote)], quote
return symbol, ""
def _asset_pair_to_standard_symbol(pair_info: dict) -> str:
"""Konvertiert Kraken AssetPairs-Metadaten zu internem Symbol."""
base = _asset_to_standard(pair_info.get("base", ""))
quote = _quote_to_standard(pair_info.get("quote", ""))
if not base or not quote:
return ""
return f"{base}{quote}"
# Spezialfälle für Kraken-Symbole die nicht ins Muster passen
_KRAKEN_EXCEPTIONS: dict[str, str] = {}
def get_ticker_24hr() -> list[dict]:
"""
Gibt alle 24h Ticker zurück (Kraken).
Returns:
Liste von Dicts mit symbol, volume, quoteVolume, etc.
"""
url = f"{config.BINANCE_BASE_URL}/0/public/Ticker"
data = _request_with_retry(url)
# Kraken Response: { error: [], result: { XXBTZUSD: { v: [...], c: [...] } } }
result = data.get("result", {})
pair_lookup = _get_asset_pair_lookup()
tickers_by_symbol: dict[str, dict] = {}
for kraken_symbol, ticker in result.items():
pair_info = pair_lookup.get(kraken_symbol)
if pair_info:
standard_symbol = _asset_pair_to_standard_symbol(pair_info)
else:
standard_symbol = _kraken_to_standard_symbol(kraken_symbol)
# Filtere Fiat-Paare und ungewöhnliche Quotes
if not standard_symbol.endswith(("USDT", "USDC", "USD", "DAI", "TUSD", "BUSD", "GUSD")):
continue
# Skip stablecoin base
base, _quote = _split_standard_symbol(standard_symbol)
if not base or base in config.STABLECOINS or base in FIAT_BASE_ASSETS:
continue
last_price = float(ticker.get("c", ["0", "0"])[0])
volume_24h_base = float(ticker.get("v", ["0", "0"])[1])
# In USD umrechnen: Kraken gibt Basiswährungs-Volumen, kein Quote-Volumen
volume_24h_usd = volume_24h_base * last_price
existing = tickers_by_symbol.get(standard_symbol)
ticker_data = {
"symbol": standard_symbol,
"lastPrice": str(last_price),
"quoteVolume": str(volume_24h_usd),
"volume": str(volume_24h_usd),
}
if existing is None or volume_24h_usd > float(existing["quoteVolume"]):
tickers_by_symbol[standard_symbol] = ticker_data
return list(tickers_by_symbol.values())
# Mapping von lesbaren Interval-Strings auf Kraken-Minuten-Werte
_INTERVAL_MAP = {
"1m": "1",
"5m": "5",
"15m": "15",
"30m": "30",
"1h": "60",
"60m": "60",
"4h": "240",
"240m": "240",
"1d": "1440",
"1w": "10080",
}
def get_klines(symbol: str, interval: str = "15", limit: int = 100) -> list:
"""
Holt OHLCV-Kerzen für ein Symbol.
Args:
symbol: z.B. "BTCUSDT" (wird zu Kraken-Format konvertiert)
interval: "15" oder "15m", "60" oder "1h", "240" oder "4h" (Minuten)
limit: Ungefähre Anzahl Kerzen — wird über `since`-Timestamp gesteuert
(Kraken unterstützt keinen count-Parameter direkt)
Returns:
Liste von [open_time, open, high, low, close, volume, close_time, ...]
(Binance-kompatibles Format)
"""
# Interval normalisieren: "15m" → "15", "4h" → "240", etc.
interval_kraken = _INTERVAL_MAP.get(
str(interval), str(interval).replace("m", "").replace("h", "")
)
interval_minutes = int(interval_kraken)
kraken_symbol = _standard_to_kraken_symbol(symbol)
# `since`-Timestamp berechnen um die Datenmenge auf ~limit Kerzen zu begrenzen
import time as _time
since = int(_time.time()) - (limit * interval_minutes * 60)
url = f"{config.BINANCE_BASE_URL}/0/public/OHLC"
params = {
"pair": kraken_symbol,
"interval": interval_kraken,
"since": since,
}
data = _request_with_retry(url, params)
# Kraken Response: { error: [], result: { XBTZUSD: [[t,o,h,l,c,vwap,v,cnt], ...], "last": 123456 } }
# Wichtig: Kraken nutzt oft einen anderen Key als den angeforderten (Altname).
# Deshalb nehmen wir den ersten Key der kein "last" ist.
result = data.get("result", {})
klines_key = next((k for k in result if k != "last"), None)
if not klines_key:
logger.warning(f"Kraken: Keine OHLC-Daten für {kraken_symbol} gefunden")
return []
klines = result[klines_key]
# Konvertiere zu Binance-kompatiblem Format
# Kraken: [time, open, high, low, close, vwap, volume, count]
# Binance: [time, open, high, low, close, volume, close_time, quote_volume, ...]
converted = []
for k in klines:
converted.append(
[
int(k[0]), # open_time
str(k[1]), # open
str(k[2]), # high
str(k[3]), # low
str(k[4]), # close
str(k[6]), # volume (Index 6 bei Kraken)
int(k[0]) + interval_minutes * 60, # close_time (geschätzt)
"0", # quote_volume (nicht direkt verfügbar)
int(k[7]), # trade count
"0", # taker_buy_base_volume
"0", # taker_buy_quote_volume
"0", # ignore
]
)
return converted
def _standard_to_kraken_symbol(symbol: str) -> str:
"""
Konvertiert Standard-Symbol zu Kraken-Format.
Beispiele:
BTCUSDT → XXBTZUSD
ETHUSDT → XETHZUSD
DOGEUSDT → XXDGZUSD
"""
target_base, target_quote = _split_standard_symbol(symbol)
if target_base and target_quote:
try:
for pair_key, pair_info in _get_asset_pair_lookup().items():
base = _asset_to_standard(pair_info.get("base", ""))
quote = _quote_to_standard(pair_info.get("quote", ""))
if base == target_base and quote == target_quote:
return str(pair_key)
except BinanceAPIError:
logger.debug("Kraken AssetPairs Lookup fehlgeschlagen, nutze statisches Mapping")
# Entferne "USDT" oder "USDC" am Ende
base = symbol.replace("USDT", "").replace("USDC", "").replace("USD", "")
# Reverse Mapping
reverse_map = {
"BTC": "XXBT",
"ETH": "XETH",
"LTC": "XLTC",
"XRP": "XXRP",
"DOGE": "XXDG",
"ADA": "XADA",
"DOT": "XDOT",
"LINK": "XLINK",
"AVAX": "XAVAX",
"MATIC": "XMATIC",
"SOL": "XSOL",
"UNI": "XUNI",
"AAVE": "XAAVE",
"ATOM": "XATOM",
"ETC": "XETC",
"NEAR": "XNEAR",
"ALGO": "XALGO",
"MANA": "XMANA",
"SAND": "XSAND",
"AXS": "XAXS",
"FTM": "XFTM",
"HBAR": "XHBAR",
"VET": "XVET",
"ICP": "XICP",
"COMP": "XCOMP",
"MKR": "XMKR",
"SNX": "XSNX",
"FI": "XFI",
"RUNE": "XRUNE",
"INJ": "XINJ",
"SEI": "XSEI",
"WIF": "XWIF",
"PEPE": "XPEPE",
"SHIB": "XSHIB",
"BLUR": "XBLUR",
"ENJ": "XENJ",
"MINA": "XMINA",
"GLMR": "XGLMR",
"MOVR": "XMOVR",
"BEAM": "XBEAM",
"CELO": "XCELO",
"ONE": "XONE",
"ONT": "XONT",
"XTZ": "XTZ",
"STORJ": "XSTORJ",
"KSM": "XKSM",
"CHZ": "XCHZ",
"ANKR": "XANKR",
"GRT": "XGRT",
"BAL": "XBAL",
"CRV": "XCRV",
"SUSHI": "XSUSHI",
"1INCH": "X1INCH",
"BNB": "XBNB",
"XBC": "XXBC",
"ZEC": "XZEC",
"DASH": "XDASH",
"BSV": "XBSV",
"TRX": "XTRX",
"EOS": "XEOS",
"NEO": "XNEO",
"IOTA": "XIOTA",
"WAVES": "XWAVES",
"LSK": "XLSK",
"STRAT": "XSTRAT",
"ARK": "XARK",
"STEEM": "XSTEEM",
"WAX": "XWAX",
"ICX": "XICX",
"IOST": "XIOST",
"REQ": "XREQ",
"POWER": "XPOWER",
"KMD": "XKMD",
"RLC": "XRLC",
"RCN": "XRCN",
"PAY": "XPAY",
"STORM": "XSTORM",
"OMG": "XOMG",
"ANT": "XANT",
"GNO": "XGNO",
"BNT": "XBNT",
"GNT": "XGNT",
"BAT": "XBAT",
"REP": "XREP",
"KNC": "XKNC",
"DAI": "XDAI",
"SAI": "XSAI",
"USDT": "XUSDT",
"USDC": "XUSDC",
"TUSD": "XTUSD",
"GUSD": "XGUSD",
"BUSD": "XBUSD",
"PAX": "XPAX",
}
kraken_base = reverse_map.get(base, f"X{base}")
# Quote: USDT → ZUSD (Kraken nutzt Z-Präfix für Fiat)
if symbol.endswith("USDT"):
return f"{kraken_base}ZUSD"
elif symbol.endswith("USDC"):
return f"{kraken_base}USDC"
elif symbol.endswith("USD"):
return f"{kraken_base}ZUSD"
elif symbol.endswith("DAI"):
return f"{kraken_base}DAI"
elif symbol.endswith("TUSD"):
return f"{kraken_base}TUSD"
elif symbol.endswith("BUSD"):
return f"{kraken_base}BUSD"
elif symbol.endswith("GUSD"):
return f"{kraken_base}GUSD"
else:
# Fallback: ZUSD
return f"{kraken_base}ZUSD"
def get_current_price(symbol: str) -> float:
"""
Gibt den aktuellen Preis eines Symbols zurück.
Args:
symbol: z.B. "BTCUSDT"
Returns:
Aktueller Preis als Float
"""
kraken_symbol = _standard_to_kraken_symbol(symbol)
url = f"{config.BINANCE_BASE_URL}/0/public/Ticker"
params = {"pair": kraken_symbol}
data = _request_with_retry(url, params)
result = data.get("result", {})
# Kraken kann unter Altname antworten statt unter dem angeforderten Key
actual_key = next(iter(result), None)
ticker = result.get(actual_key) if actual_key else None
if not ticker:
raise BinanceAPIError(f"Symbol {symbol} ({kraken_symbol}) nicht in Kraken-Tickern gefunden")
# c[0] = Last trade price
return float(ticker["c"][0])
def get_btc_4h_klines(limit: int = 50) -> list:
"""
Holt BTC 4h-Kerzen für EMA50-Berechnung und 4h-Change.
Args:
limit: Anzahl Kerzen (50 reicht für EMA50 + 4h-Change)
Returns:
Liste von OHLCV-Kerzen
"""
# Kraken Interval: "240" = 4h
return get_klines("BTCUSDT", interval="240", limit=limit)