Affiliate REST

import base64
import hashlib
import hmac
import json
import time
import urllib.error
import urllib.parse
import urllib.request

BASE_URL = "https://api.kraken.com"
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"  # base64-encoded

_last_nonce = 0


class KrakenApiError(Exception):
    """Non-2xx Kraken API response with a typed error body."""

    def __init__(self, status: int, body: dict):
        self.status = status
        self.body = body
        super().__init__(f"{status}: {body}")


def _nonce() -> str:
    global _last_nonce
    candidate = int(time.time() * 1000)
    _last_nonce = max(candidate, _last_nonce + 1)
    return str(_last_nonce)


def _flat_query(params: dict | None) -> str:
    if not params:
        return ""
    return urllib.parse.urlencode(params, doseq=True)


def _nested_query(params: dict | None) -> str:
    if not params:
        return ""
    pairs: list[tuple[str, str]] = []

    def add(name: str, value) -> None:
        if isinstance(value, dict):
            for child_name, child_value in value.items():
                add(f"{name}[{child_name}]", child_value)
        elif isinstance(value, bool):
            pairs.append((name, "true" if value else "false"))
        elif value is not None:
            pairs.append((name, str(value)))

    for key, value in params.items():
        add(key, value)
    return urllib.parse.urlencode(pairs)


def signed_rest_request(
    method: str,
    path: str,
    params: dict | None = None,
    body: dict | None = None,
    *,
    nested_query: bool = False,
):
    """Send a signed Kraken REST API request."""
    nonce = _nonce()
    query = _nested_query(params) if nested_query else _flat_query(params)
    signed_path = f"{path}?{query}" if query else path
    body_bytes = (
        json.dumps(body, separators=(",", ":")).encode() if body is not None else b""
    )
    digest = hashlib.sha256(nonce.encode() + body_bytes).digest()
    signature = base64.b64encode(
        hmac.new(
            base64.b64decode(API_SECRET),
            signed_path.encode() + digest,
            hashlib.sha512,
        ).digest()
    ).decode()

    headers = {
        "api-key": API_KEY,
        "api-sign": signature,
        "api-nonce": nonce,
    }
    if body is not None:
        headers["content-type"] = "application/json"

    request = urllib.request.Request(
        f"{BASE_URL}{signed_path}",
        data=body_bytes or None,
        headers=headers,
        method=method,
    )
    try:
        with urllib.request.urlopen(request) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as exc:
        error_body = json.loads(exc.read().decode())
        raise KrakenApiError(exc.code, error_body) from exc


def funding_request(
    method: str,
    path: str,
    params: dict | None = None,
    body: dict | None = None,
):
    """Send a signed Funding API request."""
    return signed_rest_request(method, path, params, body, nested_query=True)

page = signed_rest_request(
    "GET",
    "/affiliate/v1/daily-activity",
    {"activity_date": "2026-09-16", "limit": 50},
)
print(page["activity_date"], page["active_users"], page["currency"])
for item in page["items"]:
    print(item["referee_reference"], item["masked_iiban"])

GET /affiliate/v1/daily-activity?activity_date=2026-09-16&limit=50

page = signed_rest_request(
    "GET",
    "/affiliate/v1/daily-activity",
    {
        "iiban": "AA45N84GQK2VUN7A,BB12N84GQK2VUN7B",
        "start_date": "2026-08-20",
        "end_date": "2026-09-16",
        "limit": 50,
    },
)
for item in page["items"]:
    print(item["activity_date"], item["referee_reference"])

GET /affiliate/v1/daily-activity?iiban=AA45N84GQK2VUN7A,BB12N84GQK2VUN7B&start_date=2026-08-20&end_date=2026-09-16&limit=50