Public Market Data

class PublicClient:
    def __init__(self, api_url: str = 'https://api.pro.coinbase.com', timeout: int = 30):
        """
        Create cbpro API public client.

        Parameters:
        - api_url (str): API URL. Defaults to cbpro API.
        - timeout (int): Request timeout in seconds.
        """

import cbpro

# Use production API (default)
client = cbpro.PublicClient()

# Use sandbox API for testing
client = cbpro.PublicClient(api_url='https://api-public.sandbox.pro.coinbase.com')

def get_products(self) -> list:
    """
    Get a list of available currency pairs for trading.

    Returns:
    list: Info about all currency pairs. Each item contains:
        - id: Product identifier (e.g., "BTC-USD")
        - display_name: Human-readable name (e.g., "BTC/USD")
        - base_currency: Base currency code (e.g., "BTC")
        - quote_currency: Quote currency code (e.g., "USD")
        - base_min_size: Minimum order size for base currency
        - base_max_size: Maximum order size for base currency
        - quote_increment: Minimum price increment
    """

def get_currencies(self) -> list:
    """
    List known currencies.

    Returns:
    list: List of currencies. Each item contains:
        - id: Currency code (e.g., "BTC")
        - name: Full currency name (e.g., "Bitcoin")
        - min_size: Minimum transaction size
    """

# Get all trading pairs
products = client.get_products()
btc_products = [p for p in products if 'BTC' in p['id']]

# Get all supported currencies
currencies = client.get_currencies()
crypto_currencies = [c for c in currencies if c['id'] != 'USD']

def get_product_order_book(self, product_id: str, level: int = 1) -> dict:
    """
    Get a list of open orders for a product.

    The amount of detail shown can be customized with the level parameter:
    - 1: Only the best bid and ask
    - 2: Top 50 bids and asks (aggregated)  
    - 3: Full order book (non aggregated)

    Level 1 and Level 2 are recommended for polling. Level 3 should only be
    used with websocket streaming to avoid being rate limited.

    Parameters:
    - product_id (str): Product identifier (e.g., "BTC-USD")
    - level (int): Order book level (1, 2, or 3). Default is 1.

    Returns:
    dict: Order book data containing:
        - sequence: Message sequence number
        - bids: List of [price, size, num-orders] for buy orders
        - asks: List of [price, size, num-orders] for sell orders
    """

# Get best bid/ask only
level1 = client.get_product_order_book('BTC-USD', level=1)
best_bid = level1['bids'][0][0]  # Best buy price
best_ask = level1['asks'][0][0]  # Best sell price
spread = float(best_ask) - float(best_bid)

# Get top 50 levels for deeper analysis
level2 = client.get_product_order_book('BTC-USD', level=2)
total_bid_volume = sum(float(bid[1]) for bid in level2['bids'])

def get_product_ticker(self, product_id: str) -> dict:
    """
    Snapshot about the last trade (tick), best bid/ask and 24h volume.

    Polling is discouraged in favor of connecting via websocket stream.

    Parameters:
    - product_id (str): Product identifier

    Returns:
    dict: Ticker info containing:
        - trade_id: ID of the last trade
        - price: Last trade price
        - size: Last trade size
        - bid: Best bid price
        - ask: Best ask price
        - volume: 24-hour volume in base currency
        - time: Timestamp of last trade
    """

def get_product_24hr_stats(self, product_id: str) -> dict:
    """
    Get 24 hr stats for the product.

    Parameters:
    - product_id (str): Product identifier

    Returns:
    dict: 24 hour stats. Volume is in base currency units.
        - open: Opening price 24 hours ago
        - high: Highest price in last 24 hours
        - low: Lowest price in last 24 hours  
        - volume: Volume in base currency units
        - last: Last trade price
        - volume_30day: 30-day volume
    """

# Get current market snapshot
ticker = client.get_product_ticker('BTC-USD')
current_price = float(ticker['price'])
daily_volume = float(ticker['volume'])

# Get 24-hour statistics
stats = client.get_product_24hr_stats('BTC-USD')
price_change = float(stats['last']) - float(stats['open'])
price_change_pct = (price_change / float(stats['open'])) * 100

def get_product_trades(self, product_id: str, before: str = '', after: str = '', 
                      limit: int = None, result: list = None):
    """
    List the latest trades for a product.

    This method returns a generator which may make multiple HTTP requests
    while iterating through it.

    Parameters:
    - product_id (str): Product identifier
    - before (str): Start time in ISO 8601 format
    - after (str): End time in ISO 8601 format
    - limit (int): Maximum number of trades to return (automatically paginated)
    - result (list): Internal parameter for pagination

    Yields:
    dict: Trade information containing:
        - time: Trade timestamp in ISO 8601
        - trade_id: Unique trade identifier
        - price: Trade execution price
        - size: Trade size in base currency
        - side: Trade side ("buy" or "sell")
    """

# Get recent trades (generator)
trades = client.get_product_trades('BTC-USD')
recent_trades = list(islice(trades, 100))  # Get first 100 trades

# Calculate volume-weighted average price
total_volume = sum(float(trade['size']) for trade in recent_trades)
vwap = sum(float(trade['price']) * float(trade['size']) for trade in recent_trades) / total_volume

def get_product_historic_rates(self, product_id: str, start: str = None, end: str = None,
                              granularity: int = None) -> list:
    """
    Historic rates for a product.

    Rates are returned in grouped buckets based on requested granularity.
    Historical rate data may be incomplete. No data is published for
    intervals where there are no ticks.

    The maximum number of data points for a single request is 200 candles.
    For larger time ranges, multiple requests with different start/end times are needed.

    Parameters:
    - product_id (str): Product identifier
    - start (str): Start time in ISO 8601 format
    - end (str): End time in ISO 8601 format
    - granularity (int): Desired time slice in seconds
        Valid values: 60, 300, 900, 3600, 21600, 86400
        (1m, 5m, 15m, 1h, 6h, 1d)

    Returns:
    list: Historic candle data. Each candle is:
        [timestamp, low, high, open, close, volume]

    Raises:
    ValueError: If granularity is not in approved values
    """

from datetime import datetime, timedelta

# Get daily candles for last 30 days
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=30)

candles = client.get_product_historic_rates(
    product_id='BTC-USD',
    start=start_time.isoformat(),
    end=end_time.isoformat(),
    granularity=86400  # Daily candles
)

# Extract OHLCV data
for candle in candles:
    timestamp, low, high, open_price, close, volume = candle
    print(f"Date: {datetime.fromtimestamp(timestamp)}")
    print(f"OHLC: {open_price} / {high} / {low} / {close}")
    print(f"Volume: {volume}")

def get_time(self) -> dict:
    """
    Get the API server time.

    Returns:
    dict: Server time in both ISO and epoch format:
        - iso: Time in ISO 8601 format
        - epoch: Decimal seconds since Unix epoch
    """

# Get server time for synchronization
server_time = client.get_time()
server_timestamp = server_time['epoch']
local_timestamp = time.time()
time_drift = abs(server_timestamp - local_timestamp)

if time_drift > 30:  # More than 30 seconds drift
    print(f"Warning: Clock drift of {time_drift:.2f} seconds detected")