Bot Integration with Coinbase API

Coinbase API Bot Integration

Coinbase Advanced Trade API (formerly Coinbase Pro) — a professional API for algorithmic trading. Differs from Coinbase Simple: supports limit/market/stop orders, WebSocket feed, detailed history. Key feature — transition to JWT authentication in 2024.

Authentication via JWT

Coinbase Advanced Trade API uses JWT (JSON Web Token) with ES256 algorithm:

import jwt
import time
from cryptography.hazmat.primitives.asymmetric import ec

def generate_jwt(api_key: str, private_key_pem: str) -> str:
    """
    api_key: string like "organizations/{org_id}/apiKeys/{key_id}"
    private_key_pem: EC private key in PEM format
    """
    private_key = serialization.load_pem_private_key(
        private_key_pem.encode(), password=None
    )
    
    payload = {
        'sub': api_key,
        'iss': 'coinbase-cloud',
        'nbf': int(time.time()),
        'exp': int(time.time()) + 120,  # JWT valid for 2 minutes
    }
    
    return jwt.encode(payload, private_key, algorithm='ES256',
                      headers={'kid': api_key, 'nonce': secrets.token_hex(16)})

# Each request requires a fresh JWT
async def api_request(method: str, path: str, body: dict = None):
    token = generate_jwt(API_KEY, PRIVATE_KEY)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
    }
    async with aiohttp.ClientSession() as session:
        async with session.request(method, BASE_URL + path, 
                                    headers=headers, json=body) as resp:
            return await resp.json()

Basic Trading Operations

# Place order
async def place_order(product_id: str, side: str, order_type: str, 
                       size: str, price: str = None) -> dict:
    body = {
        'client_order_id': str(uuid.uuid4()),
        'product_id': product_id,  # 'BTC-USD'
        'side': side,               # 'BUY' or 'SELL'
        'order_configuration': {}
    }
    
    if order_type == 'market':
        if side == 'BUY':
            body['order_configuration']['market_market_ioc'] = {'quote_size': size}
        else:
            body['order_configuration']['market_market_ioc'] = {'base_size': size}
    elif order_type == 'limit':
        body['order_configuration']['limit_limit_gtc'] = {
            'base_size': size,
            'limit_price': price,
            'post_only': False,
        }
    
    return await api_request('POST', '/api/v3/brokerage/orders', body)

# Cancel orders
async def cancel_orders(order_ids: list[str]) -> dict:
    return await api_request('POST', '/api/v3/brokerage/orders/batch_cancel',
                              {'order_ids': order_ids})

WebSocket Feed

import websockets

async def subscribe_market_data():
    async with websockets.connect('wss://advanced-trade-ws.coinbase.com') as ws:
        # Subscribe with JWT authentication
        token = generate_jwt(API_KEY, PRIVATE_KEY)
        await ws.send(json.dumps({
            'type': 'subscribe',
            'product_ids': ['BTC-USD', 'ETH-USD'],
            'channel': 'ticker',
            'jwt': token,
        }))
        
        async for message in ws:
            data = json.loads(message)
            if data['channel'] == 'ticker':
                for event in data['events']:
                    process_ticker(event)

Coinbase also supports CCXT for unified access, but direct API provides more capabilities (batch cancel, advanced order types).

Coinbase Advanced Trade API integration: 1–2 weeks including JWT authentication and sandbox testing.