Prerequisites

• Completion of Crypto Lesson 1 (env & testnet setup)
• Python 3.10+, requests and python-dotenv installed
• A testnet API key & secret from your chosen exchange

Overview

This lesson covers the common HMAC signing pattern used by many exchanges, how to build and reuse a secure signed request helper, nonce/timestamp-based replay protection, and how to handle key rotation and storage.

1) Recommended .env values

Add these to your .env (never commit):

Copy# .env
EXCHANGE_API_URL=https://testnet-api.example.com
EXCHANGE_API_KEY=your_testnet_api_key
EXCHANGE_API_SECRET=your_testnet_api_secret
EXCHANGE_WS_KEY=your_ws_key (if required)
EXCHANGE_WS_SECRET=your_ws_secret (if required)

2) HMAC signing helper (auth_hmac.py)

Below is a reusable signed REST helper. Most exchanges use a similar pattern: timestamp + method + path + body → HMAC-SHA256 (or SHA512). Read your exchange docs and adapt header names and concatenation order.

Copyimport os
import time
import hmac
import hashlib
import json
import requests
from dotenv import load_dotenv

load_dotenv()

API_URL = os.getenv('EXCHANGE_API_URL').rstrip('/')
API_KEY = os.getenv('EXCHANGE_API_KEY')
API_SECRET = os.getenv('EXCHANGE_API_SECRET').encode()

def sign_message(message: bytes) -> str:
    # HMAC-SHA256 hex digest
    return hmac.new(API_SECRET, message, hashlib.sha256).hexdigest()

def make_signed_headers(method: str, path: str, body: dict=None):
    ts = str(int(time.time() * 1000))  # milliseconds
    body_json = json.dumps(body) if body else ''
    # Most exchanges use: ts + method + path + body
    prehash = (ts + method.upper() + path + body_json).encode()
    signature = sign_message(prehash)
    return {
        'API-KEY': API_KEY,
        'API-SIGN': signature,
        'API-TIMESTAMP': ts,
        'Content-Type': 'application/json'
    }

def signed_request(method: str, path: str, params=None, json_body=None, timeout=10):
    url = API_URL + path
    headers = make_signed_headers(method, path, json_body)
    r = requests.request(method.upper(), url, headers=headers, params=params, json=json_body, timeout=timeout)
    r.raise_for_status()
    return r.json()

# Example usage:
if __name__ == '__main__':
    print('Markets:', signed_request('GET', '/api/v1/markets'))

Adaptation note: Some exchanges SHA256 the body separately, some include query string ordering, some use API key header names like X-MBX-APIKEY (Binance). Always match the provider spec.

3) Replay protection & nonces

Use timestamp windows and/or nonces so servers can reject replayed requests. On the client side:

• Include a millisecond timestamp and sign it (as above).
• Reject server responses with “timestamp skew” errors — sync clock via NTP.
• Optionally include an incrementing nonce stored in a small file for persistent ordering.

Copy# Example nonce file helper
from pathlib import Path
def load_nonce(path='.nonce'):
    p = Path(path)
    if not p.exists(): 
        p.write_text('0')
    n = int(p.read_text().strip() or '0') + 1
    p.write_text(str(n))
    return n

4) WebSocket auth (if required)

Some exchanges require signed WS auth. A typical pattern: compute a signature over a CONNECT string or timestamp and send an auth message after connecting.

Copyimport asyncio, hmac, hashlib, time, json, os
import websockets
from dotenv import load_dotenv
load_dotenv()

WS_URL = os.getenv('EXCHANGE_WS_URL')
API_KEY = os.getenv('EXCHANGE_API_KEY')
API_SECRET = os.getenv('EXCHANGE_API_SECRET').encode()

def ws_sign(ts):
    msg = f"{ts}GET/realtime"
    return hmac.new(API_SECRET, msg.encode(), hashlib.sha256).hexdigest()

async def ws_auth_example():
    async with websockets.connect(WS_URL) as ws:
        ts = str(int(time.time() * 1000))
        sig = ws_sign(ts)
        auth_msg = {'op':'auth','args': [API_KEY, ts, sig]}
        await ws.send(json.dumps(auth_msg))
        resp = await ws.recv()
        print('auth response', resp)

WS auth formats differ. Some exchanges send API key in URL query, others require a JSON auth message. Read the exchange docs.

5) Retry, backoff & error handling

Wrap signed_request with retries for transient network errors and 5xx responses. Do not automatically retry 4xx errors (bad signature, insufficient funds).

Copyimport time
from requests import RequestException

def resilient_request(method, path, retries=3, backoff=0.5, **kwargs):
    for attempt in range(1, retries+1):
        try:
            return signed_request(method, path, **kwargs)
        except RequestException as e:
            status = getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None
            # retry on 5xx or network errors
            if attempt < retries and (status is None or 500 <= status < 600):
                time.sleep(backoff * (2 ** (attempt-1)))
                continue
            raise

6) Key management & rotation

• Keep API keys in environment variables or a secrets manager (AWS Secrets Manager, Vault).
• Rotate keys periodically and automate invalidation of old keys.
• Limit key permissions (read-only vs trade) for lower risk during development.
• Use file permissions (chmod 600) for any local files containing secrets.

7) Example: place a small test order (testnet)

Use the signed helper to place a tiny test order on the exchange testnet. Replace endpoint and request body to match the exchange API.

Copy# test_order.py
from auth_hmac import resilient_request
order = {
  "symbol": "BTC/USDT",
  "side": "BUY",
  "type": "LIMIT",
  "price": 20000.0,
  "quantity": 0.001
}
resp = resilient_request('POST', '/api/v1/orders', json_body=order)
print('Order response:', resp)

Always test in sandbox/testnet and use tiny sizes when moving to live accounts.

What you'll build next

Lesson 3 will implement market data ingestion via REST and WebSocket (order book snapshots and deltas) and show normalization patterns to unify different exchanges.

Back to course hub