133 lines
5.5 KiB
Python
133 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import logging
|
|
import time
|
|
import urllib.parse
|
|
|
|
import requests
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_BASE_URL = "https://api.kraken.com"
|
|
|
|
|
|
class KrakenError(Exception):
|
|
pass
|
|
|
|
|
|
class KrakenClient:
|
|
def __init__(self, api_key: str = "", api_secret: str = ""):
|
|
self.api_key = api_key
|
|
self.api_secret = api_secret
|
|
self._session = requests.Session()
|
|
self._session.headers["User-Agent"] = "crypto-trader/1.0"
|
|
|
|
# ── Auth ─────────────────────────────────────────────────────────────────
|
|
|
|
def _sign(self, urlpath: str, data: dict) -> str:
|
|
postdata = urllib.parse.urlencode(data)
|
|
encoded = (str(data["nonce"]) + postdata).encode()
|
|
message = urlpath.encode() + hashlib.sha256(encoded).digest()
|
|
mac = hmac.new(base64.b64decode(self.api_secret), message, hashlib.sha512)
|
|
return base64.b64encode(mac.digest()).decode()
|
|
|
|
# ── Low-level request helpers ─────────────────────────────────────────────
|
|
|
|
def _public(self, endpoint: str, params: dict | None = None) -> dict:
|
|
url = f"{_BASE_URL}/0/public/{endpoint}"
|
|
resp = self._session.get(url, params=params, timeout=15)
|
|
resp.raise_for_status()
|
|
body = resp.json()
|
|
if body.get("error"):
|
|
raise KrakenError(body["error"])
|
|
return body["result"]
|
|
|
|
def _private(self, endpoint: str, data: dict | None = None) -> dict:
|
|
if not self.api_key or not self.api_secret:
|
|
raise KrakenError("API credentials not set — check KRAKEN_API_KEY / KRAKEN_API_SECRET")
|
|
urlpath = f"/0/private/{endpoint}"
|
|
payload = dict(data or {})
|
|
payload["nonce"] = str(int(time.time() * 1000))
|
|
headers = {
|
|
"API-Key": self.api_key,
|
|
"API-Sign": self._sign(urlpath, payload),
|
|
}
|
|
resp = self._session.post(
|
|
f"{_BASE_URL}{urlpath}", data=payload, headers=headers, timeout=15
|
|
)
|
|
resp.raise_for_status()
|
|
body = resp.json()
|
|
if body.get("error"):
|
|
raise KrakenError(body["error"])
|
|
return body["result"]
|
|
|
|
# ── Public market data ────────────────────────────────────────────────────
|
|
|
|
def get_asset_pairs(self) -> dict[str, dict]:
|
|
"""Return all asset pair metadata keyed by Kraken's internal pair name."""
|
|
return self._public("AssetPairs")
|
|
|
|
def get_tickers(self, pairs: list[str]) -> dict[str, dict]:
|
|
"""
|
|
Fetch ticker data for a list of pair names (altnames or internal names).
|
|
Splits into chunks of 100 to stay within API limits.
|
|
Returns a dict keyed by whatever name Kraken echoes back.
|
|
"""
|
|
results: dict[str, dict] = {}
|
|
for i in range(0, len(pairs), 100):
|
|
chunk = pairs[i : i + 100]
|
|
data = self._public("Ticker", params={"pair": ",".join(chunk)})
|
|
results.update(data)
|
|
return results
|
|
|
|
# ── Private account data ──────────────────────────────────────────────────
|
|
|
|
def get_all_balances(self) -> dict[str, float]:
|
|
"""Return all asset balances with non-zero amounts."""
|
|
raw = self._private("Balance")
|
|
return {k: float(v) for k, v in raw.items() if float(v) > 0}
|
|
|
|
def get_usd_balance(self) -> float:
|
|
"""Return available USD balance (ZUSD key in Kraken)."""
|
|
balance = self._private("Balance")
|
|
return float(balance.get("ZUSD", balance.get("USD", 0.0)))
|
|
|
|
# ── Order placement ───────────────────────────────────────────────────────
|
|
|
|
def market_buy(self, pair: str, volume: float, paper: bool = True) -> str:
|
|
"""Place a market buy order. Returns a transaction/order ID."""
|
|
if paper:
|
|
order_id = f"PAPER-BUY-{pair}-{int(time.time())}"
|
|
log.info("[PAPER] BUY %s qty=%.8f order=%s", pair, volume, order_id)
|
|
return order_id
|
|
result = self._private("AddOrder", {
|
|
"pair": pair,
|
|
"type": "buy",
|
|
"ordertype": "market",
|
|
"volume": f"{volume:.8f}",
|
|
})
|
|
txids = result.get("txid", [])
|
|
order_id = txids[0] if txids else "unknown"
|
|
log.info("BUY order placed: %s pair=%s qty=%.8f", order_id, pair, volume)
|
|
return order_id
|
|
|
|
def market_sell(self, pair: str, volume: float, paper: bool = True) -> str:
|
|
"""Place a market sell order. Returns a transaction/order ID."""
|
|
if paper:
|
|
order_id = f"PAPER-SELL-{pair}-{int(time.time())}"
|
|
log.info("[PAPER] SELL %s qty=%.8f order=%s", pair, volume, order_id)
|
|
return order_id
|
|
result = self._private("AddOrder", {
|
|
"pair": pair,
|
|
"type": "sell",
|
|
"ordertype": "market",
|
|
"volume": f"{volume:.8f}",
|
|
})
|
|
txids = result.get("txid", [])
|
|
order_id = txids[0] if txids else "unknown"
|
|
log.info("SELL order placed: %s pair=%s qty=%.8f", order_id, pair, volume)
|
|
return order_id
|