Kalshi API Getting Started: Python, Order Books & Authentication

#!/usr/bin/env python3
"""Kalshi public data quickstart. Python 3.10+, no dependencies or credentials.
python3 kalshi_quickstart.py --series KXHIGHNY --pages 2
At most five market-list pages, one order book and one historical cutoff request.
No account access, signing, order submission, retries or background polling.
"""
import argparse
import json
import sys
from decimal import Decimal, InvalidOperation
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, quote
from urllib.request import Request, urlopen

BASE = 'https://external-api.kalshi.com/trade-api/v2'


def get_json(route, params=None):
    url = BASE + route + ('?' + urlencode(params) if params else '')
    request = Request(url, headers={'Accept': 'application/json', 'User-Agent': 'PMTools-quickstart/1.0'})
    try:
        with urlopen(request, timeout=15) as response:
            data = json.load(response)
    except HTTPError as error:
        if error.code == 429:
            raise ValueError('Rate limited; Retry-After: ' + str(error.headers.get('Retry-After', 'not supplied')) + '. Stop and wait before retrying.') from error
        raise ValueError(f'HTTP {error.code} from {route}; check the current API documentation.') from error
    except (URLError, TimeoutError, json.JSONDecodeError) as error:
        raise ValueError(f'Network or JSON error: {error}') from error
    if not isinstance(data, dict):
        raise ValueError('Expected a JSON object; response schema may have changed.')
    return data


def best_bid(levels):
    """Do not rely on array ordering or turn an empty book into a zero quote."""
    if not isinstance(levels, list):
        raise ValueError('Expected an array of order-book levels.')
    parsed = []
    for level in levels:
        if not isinstance(level, list) or len(level) != 2:
            raise ValueError('Expected [price_dollars, count_fp].')
        price, size = (Decimal(str(value)) for value in level)
        if not price.is_finite() or not size.is_finite() or not 0 <= price <= 1 or size < 0:
            raise ValueError('Invalid order-book price or size.')
        if size > 0:
            parsed.append((price, size))
    return max(parsed, key=lambda level: level[0]) if parsed else None


def summarize_book(data):
    book = data['orderbook_fp']
    yes = best_bid(book['yes_dollars'])
    no = best_bid(book['no_dollars'])
    ask = Decimal('1') - no[0] if no else None
    return {
        'yes_bid_dollars': str(yes[0]) if yes else None,
        'yes_bid_contracts': str(yes[1]) if yes else None,
        'yes_ask_dollars': str(ask) if ask is not None else None,
        'yes_ask_contracts': str(no[1]) if no else None,
        'spread_dollars': str(ask - yes[0]) if ask is not None and yes else None,
        'note': 'YES ask = $1 minus best NO bid. Snapshot quantities are not guaranteed fills.',
    }


def collect(series, pages, request=get_json):
    cursor, seen_cursors, seen_tickers, rows = '', set(), set(), []
    for _ in range(pages):
        params = {'series_ticker': series, 'status': 'open', 'limit': 5}
        if cursor:
            params['cursor'] = cursor
        data = request('/markets', params)
        if not isinstance(data.get('markets'), list):
            raise ValueError('Missing markets array.')
        for market in data['markets']:
            ticker = market['ticker']
            if ticker not in seen_tickers:
                seen_tickers.add(ticker)
                rows.append({key: market.get(key) for key in ['ticker', 'title', 'status', 'yes_bid_dollars', 'yes_ask_dollars', 'volume_fp']})
        cursor = data.get('cursor') or ''
        if not isinstance(cursor, str):
            raise ValueError('Expected a string cursor.')
        if not cursor:
            break
        if cursor in seen_cursors:
            raise ValueError('Repeated cursor; stopping to avoid a pagination loop.')
        seen_cursors.add(cursor)
    result = {'markets': rows, 'more_available': bool(cursor), 'historical_cutoff': request('/historical/cutoff')}
    if rows:
        ticker = rows[0]['ticker']
        result['orderbook'] = {'ticker': ticker, **summarize_book(request('/markets/' + quote(ticker, safe='') + '/orderbook'))}
    else:
        result['note'] = 'No open markets for this series. Choose a current series; empty does not mean zero price.'
    return result


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--series', default='KXHIGHNY')
    parser.add_argument('--pages', type=int, choices=range(1, 6), default=2)
    args = parser.parse_args()
    try:
        print(json.dumps(collect(args.series, args.pages), indent=2))
    except (ValueError, InvalidOperation, KeyError, TypeError) as error:
        print('Stopped: ' + str(error), file=sys.stderr)
        sys.exit(1)