or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authenticated-client.mdindex.mdorder-book.mdpublic-client.mdwebsocket-client.md
tile.json

tessl/pypi-cbpro

The unofficial Python client for the Coinbase Pro API providing comprehensive trading and market data access

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/cbpro@1.1.x

To install, run

npx @tessl/cli install tessl/pypi-cbpro@1.1.0

index.mddocs/

CBPro

The unofficial Python client for the Coinbase Pro API providing comprehensive access to cryptocurrency trading, market data, and account management. This library enables developers to build automated trading systems, portfolio management tools, and market analysis applications with full access to Coinbase Pro's trading infrastructure.

Package Information

  • Package Name: cbpro
  • Language: Python
  • Installation: pip install cbpro

Core Imports

import cbpro

Direct class imports:

from cbpro import PublicClient, AuthenticatedClient, WebsocketClient, OrderBook, CBProAuth

Basic Usage

import cbpro

# Public market data access
public_client = cbpro.PublicClient()

# Get all available trading pairs
products = public_client.get_products()
print(f"Available products: {len(products)}")

# Get current price for BTC-USD
ticker = public_client.get_product_ticker('BTC-USD')
print(f"BTC-USD price: ${ticker['price']}")

# Get order book
order_book = public_client.get_product_order_book('BTC-USD', level=1)
print(f"Best bid: ${order_book['bids'][0][0]}")
print(f"Best ask: ${order_book['asks'][0][0]}")

# Authenticated trading (requires API credentials)
auth_client = cbpro.AuthenticatedClient(key, b64secret, passphrase)

# Get account balances
accounts = auth_client.get_accounts()
for account in accounts:
    if float(account['balance']) > 0:
        print(f"{account['currency']}: {account['balance']}")

# Place a limit buy order
order = auth_client.place_limit_order(
    product_id='BTC-USD',
    side='buy',
    price='30000.00',
    size='0.001'
)
print(f"Order placed: {order['id']}")

Architecture

CBPro is organized around five main components that provide comprehensive access to Coinbase Pro's API:

  • PublicClient: Access to public market data, product information, and historical data without authentication
  • AuthenticatedClient: Full trading functionality including order management, account access, and private data (inherits from PublicClient)
  • WebsocketClient: Real-time streaming of market data, order updates, and account changes via WebSocket connections
  • OrderBook: Automated maintenance of real-time order book state from WebSocket messages
  • CBProAuth: Authentication handler for securing API requests with HMAC-SHA256 signatures

This design enables everything from simple market data queries to sophisticated trading systems with real-time data processing and automated order book maintenance.

Capabilities

Public Market Data

Access to all public Coinbase Pro endpoints including products, order books, trade history, candlestick data, and market statistics. No authentication required.

class PublicClient:
    def __init__(self, api_url: str = 'https://api.pro.coinbase.com', timeout: int = 30): ...
    def get_products(self) -> list: ...
    def get_product_order_book(self, product_id: str, level: int = 1) -> dict: ...
    def get_product_ticker(self, product_id: str) -> dict: ...
    def get_product_trades(self, product_id: str, before: str = '', after: str = '', limit: int = None, result: list = None): ...
    def get_product_historic_rates(self, product_id: str, start: str = None, end: str = None, granularity: int = None) -> list: ...
    def get_product_24hr_stats(self, product_id: str) -> dict: ...
    def get_currencies(self) -> list: ...
    def get_time(self) -> dict: ...

Public Market Data

Authenticated Trading

Complete trading functionality including order placement, cancellation, account management, funding operations, and private data access. Requires API credentials.

class AuthenticatedClient(PublicClient):
    def __init__(self, key: str, b64secret: str, passphrase: str, api_url: str = "https://api.pro.coinbase.com"): ...
    def get_accounts(self) -> list: ...
    def get_account(self, account_id: str) -> dict: ...
    def get_account_history(self, account_id: str, **kwargs): ...
    def place_limit_order(self, product_id: str, side: str, price: str, size: str, **kwargs) -> dict: ...
    def place_market_order(self, product_id: str, side: str, size: str = None, funds: str = None, **kwargs) -> dict: ...
    def place_stop_order(self, product_id: str, stop_type: str, price: str, **kwargs) -> dict: ...
    def cancel_order(self, order_id: str) -> list: ...
    def cancel_all(self, product_id: str = None) -> list: ...
    def get_orders(self, product_id: str = None, status: str = None, **kwargs): ...
    def get_fills(self, product_id: str = None, order_id: str = None, **kwargs): ...
    def deposit(self, amount: str, currency: str, payment_method_id: str) -> dict: ...
    def coinbase_deposit(self, amount: str, currency: str, coinbase_account_id: str) -> dict: ...
    def withdraw(self, amount: str, currency: str, payment_method_id: str) -> dict: ...
    def coinbase_withdraw(self, amount: str, currency: str, coinbase_account_id: str) -> dict: ...
    def crypto_withdraw(self, amount: str, currency: str, crypto_address: str) -> dict: ...
    def get_fees(self) -> dict: ...

Authenticated Trading

Real-time Data Streaming

WebSocket client for real-time market data, order updates, and account changes with customizable message handling and optional MongoDB integration.

class WebsocketClient:
    def __init__(self, url: str = "wss://ws-feed.pro.coinbase.com", products: list = None, 
                 message_type: str = "subscribe", mongo_collection = None, should_print: bool = True,
                 auth: bool = False, api_key: str = "", api_secret: str = "", api_passphrase: str = "",
                 *, channels: list): ...
    def start(self): ...
    def close(self): ...
    def on_open(self): ...
    def on_message(self, msg: dict): ...
    def on_close(self): ...
    def on_error(self, e: Exception, data = None): ...

Real-time Data Streaming

Order Book Management

Real-time order book maintenance that automatically processes WebSocket messages to maintain accurate bid/ask state for a specific product.

class OrderBook:
    def __init__(self, product_id: str = 'BTC-USD', log_to = None): ...
    def process_message(self, message: dict): ...
    def get_current_book(self) -> dict: ...
    def get_ask(self): ...
    def get_bid(self): ...
    def get_current_ticker(self) -> dict: ...

Order Book Management

Types

class CBProAuth:
    """Authentication handler for API requests."""
    def __init__(self, api_key: str, secret_key: str, passphrase: str): ...
    def __call__(self, request): ...

def get_auth_headers(timestamp: str, message: str, api_key: str, secret_key: str, passphrase: str) -> dict:
    """Generate authentication headers for API requests."""

Error Handling

The library handles various error conditions:

  • API Rate Limits: Automatic retry logic for rate-limited requests
  • Network Errors: Connection timeout handling and retry mechanisms
  • Authentication Errors: Clear error messages for invalid credentials
  • WebSocket Disconnections: Automatic reconnection and error recovery
  • Order Book Gaps: Automatic resyncing when message sequence gaps are detected

Environment Support

  • Production: https://api.pro.coinbase.com (default)
  • Sandbox: https://api-public.sandbox.pro.coinbase.com for testing