Disclosure: PredScope may receive compensation when you sign up for prediction market platforms through links on this site. This does not influence our ratings or reviews. Learn more.

Home › Guides › Kalshi API

Kalshi API Guide 2026: Endpoints, Authentication & Code Examples

Updated March 2026 — The complete developer guide to the Kalshi trading API. Learn how to authenticate, query markets, place trades, stream real-time data via websockets, and build automated trading bots with Python.

The Kalshi API gives developers and algorithmic traders programmatic access to the CFTC-regulated prediction market exchange. Whether you want to pull market data for analysis, build a trading bot, or integrate prediction market prices into your application, the Kalshi REST API and websocket feeds provide everything you need.

This guide covers the complete Kalshi API documentation you need to get started: authentication, every major endpoint category, Python code examples, websocket streaming, rate limits, error handling, and a full trading bot example. If you are new to Kalshi, start with our How Does Kalshi Work guide first.

API Overview & Architecture

The Kalshi trading API is a RESTful HTTP API that follows standard conventions. All requests and responses use JSON. The API is organized into several major categories:

API Environments

Kalshi provides two separate environments for API access:

Getting Started: API Key Setup

Before you can make authenticated requests to the Kalshi REST API, you need to generate an API key pair from your Kalshi account. Here is the step-by-step process:

1. Create a Kalshi account.

   If you do not have one yet, sign up for Kalshi here. You will need to complete identity verification (KYC) before you can trade.

2. Navigate to API Key Settings.

   Log in to your Kalshi account, go to Settings → API Keys, and click "Generate New Key".

3. Generate an RSA key pair.

   Kalshi uses RSA key pairs for API authentication. You can generate a key pair using OpenSSL on your command line:

# Generate a 4096-bit RSA private key
openssl genrsa -out kalshi_private_key.pem 4096

# Extract the public key from the private key
openssl rsa -in kalshi_private_key.pem -pubout -out kalshi_public_key.pem

# View the public key (you will paste this into Kalshi)
cat kalshi_public_key.pem

4. Upload your public key to Kalshi.

   Paste the contents of kalshi_public_key.pem into the Kalshi API key creation form. Kalshi will give you a Key ID (a UUID string). Save this Key ID — you will need it for every API request.

5. Store your private key securely.

   Your private key (kalshi_private_key.pem) must be kept secret. Never commit it to version control, share it publicly, or include it in client-side code.

Install Required Python Libraries

For the Python examples in this guide, install these dependencies:

# Core dependencies
pip install requests cryptography websockets

# Optional: Kalshi's official Python SDK
pip install kalshi-python

Authentication & Request Signing

The Kalshi API authentication system uses RSA key signatures. For every authenticated request, you must generate a signature from the request timestamp, HTTP method, and path, then include it in your request headers.

Authentication Headers

Every authenticated request requires three custom headers:

Python Authentication Implementation

Here is a complete Python class that handles Kalshi API authentication and request signing:

import time
import base64
import requests
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding


class KalshiClient:
    """Client for the Kalshi trading API with RSA authentication."""

    def __init__(self, key_id: str, private_key_path: str, base_url: str = None):
        self.key_id = key_id
        self.base_url = base_url or "https://api.elections.kalshi.com/trade-api/v2"
        self.session = requests.Session()

        # Load the RSA private key from file
        with open(private_key_path, "rb") as f:
            self.private_key = serialization.load_pem_private_key(
                f.read(), password=None
            )

    def _get_auth_headers(self, method: str, path: str) -> dict:
        """Generate authentication headers for a request."""
        # Current timestamp in milliseconds
        timestamp = str(int(time.time() * 1000))

        # Build the message to sign: timestamp + method + path
        message = timestamp + method.upper() + path

        # Sign with RSA-PKCS1v15-SHA256
        signature = self.private_key.sign(
            message.encode("utf-8"),
            padding.PKCS1v15(),
            hashes.SHA256()
        )

        return {
            "KALSHI-ACCESS-KEY": self.key_id,
            "KALSHI-ACCESS-SIGNATURE": base64.b64encode(signature).decode(),
            "KALSHI-ACCESS-TIMESTAMP": timestamp,
            "Content-Type": "application/json",
        }

    def get(self, path: str, params: dict = None) -> dict:
        """Make an authenticated GET request."""
        url = self.base_url + path
        headers = self._get_auth_headers("GET", path)
        response = self.session.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()

    def post(self, path: str, data: dict = None) -> dict:
        """Make an authenticated POST request."""
        url = self.base_url + path
        headers = self._get_auth_headers("POST", path)
        response = self.session.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()

    def delete(self, path: str) -> dict:
        """Make an authenticated DELETE request."""
        url = self.base_url + path
        headers = self._get_auth_headers("DELETE", path)
        response = self.session.delete(url, headers=headers)
        response.raise_for_status()
        return response.json()


# Initialize the client
client = KalshiClient(
    key_id="your-api-key-id-here",
    private_key_path="kalshi_private_key.pem",
    # Use demo for testing:
    # base_url="https://demo-api.kalshi.co/trade-api/v2"
)

Market & Event Endpoints

Market data endpoints let you browse events, query market details, fetch orderbooks, and retrieve historical trade data. Most market data endpoints do not require authentication, making them ideal for building dashboards or analysis tools.

Get Exchange Status

Check whether the exchange is currently open for trading:

# Check exchange status
status = client.get("/exchange/status")
print(status)
# {
#   "exchange_active": true,
#   "trading_active": true
# }

List Events

Events are the top-level containers on Kalshi. Each event (e.g., "Fed Rate Decision March 2026") contains one or more markets (e.g., "Will the Fed raise rates?"):

# List events with optional filters
events = client.get("/events", params={
    "status": "open",           # open, closed, settled
    "series_ticker": "FED",     # filter by series
    "limit": 20,                # results per page (max 200)
    "cursor": None              # pagination cursor
})

for event in events["events"]:
    print(f"{event['event_ticker']}: {event['title']}")
    print(f"  Category: {event['category']}")
    print(f"  Markets: {event['markets_count']}")
    print()

Get Market Details

Retrieve full details for a specific market, including current prices, volume, and status:

# Get a specific market by ticker
market = client.get("/markets/KXBTCD-26MAR31-T55000")
m = market["market"]

print(f"Title: {m['title']}")
print(f"Ticker: {m['ticker']}")
print(f"Status: {m['status']}")
print(f"Yes Bid: ${m['yes_bid'] / 100:.2f}")
print(f"Yes Ask: ${m['yes_ask'] / 100:.2f}")
print(f"Last Price: ${m['last_price'] / 100:.2f}")
print(f"Volume: {m['volume']} contracts")
print(f"Open Interest: {m['open_interest']} contracts")
print(f"Expiration: {m['expiration_time']}")

List Markets with Filters

Search and filter across all available markets:

# List markets with various filters
markets = client.get("/markets", params={
    "event_ticker": "KXBTC-26MAR31",  # filter by event
    "status": "open",                   # open, closed, settled
    "series_ticker": "KXBTC",          # filter by series
    "tickers": "KXBTCD-26MAR31-T55000,KXBTCD-26MAR31-T60000",  # specific tickers
    "limit": 100,
    "cursor": None
})

for m in markets["markets"]:
    yes_price = m["yes_bid"] / 100 if m["yes_bid"] else 0
    print(f"{m['ticker']}: {m['title']} - Yes: ${yes_price:.2f}")

Get Orderbook

Retrieve the current orderbook (bids and asks) for a market:

# Get orderbook for a market
orderbook = client.get("/markets/KXBTCD-26MAR31-T55000/orderbook")
ob = orderbook["orderbook"]

print("=== YES Bids ===")
for level in ob.get("yes", []):
    print(f"  ${level[0] / 100:.2f} x {level[1]} contracts")

print("\n=== NO Bids ===")
for level in ob.get("no", []):
    print(f"  ${level[0] / 100:.2f} x {level[1]} contracts")

Get Market Trade History

Retrieve historical trades for a specific market:

# Get recent trades for a market
trades = client.get("/markets/KXBTCD-26MAR31-T55000/trades", params={
    "limit": 50,                # max trades to return
    "cursor": None              # pagination cursor
})

for trade in trades["trades"]:
    side = "Yes" if trade["taker_side"] == "yes" else "No"
    price = trade["yes_price"] / 100
    print(f"{trade['created_time']}: {side} @ ${price:.2f} x {trade['count']}")

Get Candlestick (OHLC) Data

Retrieve candlestick data for charting and technical analysis:

# Get candlestick data for a market
candles = client.get("/markets/KXBTCD-26MAR31-T55000/candlesticks", params={
    "series_ticker": "KXBTC",
    "period_interval": 60,      # interval in minutes (1, 5, 15, 60, 1440)
})

for candle in candles.get("candlesticks", []):
    print(f"Time: {candle['end_period_ts']}")
    print(f"  Open: {candle['price']['open']}  High: {candle['price']['high']}")
    print(f"  Low: {candle['price']['low']}    Close: {candle['price']['close']}")
    print(f"  Volume: {candle['volume']}")
    print()

Trading Endpoints (Orders & Positions)

Trading endpoints let you place, amend, and cancel orders on Kalshi markets. All trading endpoints require authentication. Remember that every executed trade incurs a 2-cent per contract trading fee.

Place an Order

Submit a new order to buy or sell contracts:

# Place a limit order to buy 10 Yes contracts at $0.45
order = client.post("/portfolio/orders", data={
    "ticker": "KXBTCD-26MAR31-T55000",
    "action": "buy",            # buy or sell
    "side": "yes",              # yes or no
    "type": "limit",            # limit or market
    "count": 10,                # number of contracts
    "yes_price": 45,            # price in cents (for limit orders)
    "expiration_ts": None,      # optional: auto-cancel timestamp
    "client_order_id": "my-order-001"  # optional: your custom ID
})

print(f"Order ID: {order['order']['order_id']}")
print(f"Status: {order['order']['status']}")
print(f"Remaining: {order['order']['remaining_count']}")

Order Types

Place a Market Order

# Place a market order to buy 5 Yes contracts at best price
order = client.post("/portfolio/orders", data={
    "ticker": "KXBTCD-26MAR31-T55000",
    "action": "buy",
    "side": "yes",
    "type": "market",
    "count": 5,
})

print(f"Order filled at avg price: ${order['order']['avg_price'] / 100:.2f}")

Amend an Order

Modify an existing open order's price or quantity:

# Amend an existing order (change price and/or count)
amended = client.post(f"/portfolio/orders/{order_id}/amend", data={
    "count": 15,          # new contract count
    "yes_price": 48,      # new price in cents
})

print(f"Amended order: {amended['order']['order_id']}")

Cancel an Order

# Cancel a specific order
result = client.delete(f"/portfolio/orders/{order_id}")
print(f"Cancelled: {result['order']['order_id']}")

# Cancel all open orders (use with caution)
# This is done by retrieving and cancelling each order

Batch Order Operations

For efficiency, the Kalshi API supports batch order creation and cancellation:

# Create multiple orders in a single request
batch = client.post("/portfolio/orders/batched", data={
    "orders": [
        {
            "ticker": "KXBTCD-26MAR31-T55000",
            "action": "buy",
            "side": "yes",
            "type": "limit",
            "count": 10,
            "yes_price": 40,
        },
        {
            "ticker": "KXBTCD-26MAR31-T55000",
            "action": "buy",
            "side": "no",
            "type": "limit",
            "count": 10,
            "yes_price": 60,  # buying No at 60c = selling Yes at 40c
        }
    ]
})

for order in batch["orders"]:
    print(f"Order {order['order_id']}: {order['status']}")

Portfolio & Account Endpoints

Portfolio endpoints let you check your balance, view positions, review order history, and track fills. All require authentication.

Get Account Balance

# Get your current account balance
balance = client.get("/portfolio/balance")
print(f"Available balance: ${balance['balance'] / 100:.2f}")
print(f"Bonus balance: ${balance.get('bonus_balance', 0) / 100:.2f}")

Get Current Positions

# List all your current positions
positions = client.get("/portfolio/positions", params={
    "status": "open",           # open or settled
    "limit": 100,
    "cursor": None,
    "settlement_status": None,  # all, settled, unsettled
})

for pos in positions["market_positions"]:
    ticker = pos["ticker"]
    yes_count = pos.get("position", 0)
    avg_cost = pos.get("total_cost", 0)
    print(f"{ticker}: {yes_count} contracts, cost: ${avg_cost / 100:.2f}")

Get Order History

# Get your order history
orders = client.get("/portfolio/orders", params={
    "ticker": None,             # filter by market ticker
    "status": None,             # resting, canceled, filled, etc.
    "limit": 50,
    "cursor": None
})

for order in orders["orders"]:
    print(f"{order['created_time']}: {order['action']} {order['side']}")
    print(f"  Ticker: {order['ticker']}")
    print(f"  Count: {order['count']} @ ${order.get('yes_price', 0) / 100:.2f}")
    print(f"  Status: {order['status']}")
    print()

Get Trade Fills

# Get your fill history (executed trades)
fills = client.get("/portfolio/fills", params={
    "ticker": None,             # filter by market
    "order_id": None,           # filter by specific order
    "limit": 50,
    "cursor": None,
})

for fill in fills["fills"]:
    print(f"{fill['created_time']}: {fill['action']} {fill['side']}")
    print(f"  {fill['count']} contracts @ ${fill['yes_price'] / 100:.2f}")
    print(f"  Fee: ${fill.get('fee', 0) / 100:.2f}")
    print()

Get Settlement History

# Get your settlement history (resolved positions)
settlements = client.get("/portfolio/settlements", params={
    "limit": 50,
    "cursor": None
})

for s in settlements["settlements"]:
    pnl = s.get("revenue", 0) - s.get("total_cost", 0)
    print(f"{s['ticker']}: {'Won' if pnl > 0 else 'Lost'}")
    print(f"  Revenue: ${s.get('revenue', 0) / 100:.2f}")
    print(f"  P&L: ${pnl / 100:.2f}")
    print()

WebSocket Streaming API

The Kalshi websocket API provides real-time streaming data with lower latency than polling REST endpoints. This is essential for algorithmic trading, market making, and building live dashboards.

WebSocket Connection URLs

Available Channels

WebSocket Connection Example (Python)

import asyncio
import json
import time
import base64
import websockets
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding


async def connect_kalshi_ws(key_id: str, private_key_path: str):
    """Connect to Kalshi WebSocket with authentication."""

    # Load private key
    with open(private_key_path, "rb") as f:
        private_key = serialization.load_pem_private_key(f.read(), password=None)

    # Generate auth headers for websocket
    timestamp = str(int(time.time() * 1000))
    method = "GET"
    path = "/trade-api/ws/v2"
    message = timestamp + method + path

    signature = private_key.sign(
        message.encode("utf-8"),
        padding.PKCS1v15(),
        hashes.SHA256()
    )

    ws_url = "wss://api.elections.kalshi.com/trade-api/ws/v2"

    # Connect with auth headers
    headers = {
        "KALSHI-ACCESS-KEY": key_id,
        "KALSHI-ACCESS-SIGNATURE": base64.b64encode(signature).decode(),
        "KALSHI-ACCESS-TIMESTAMP": timestamp,
    }

    async with websockets.connect(ws_url, extra_headers=headers) as ws:
        # Subscribe to orderbook updates for a market
        await ws.send(json.dumps({
            "id": 1,
            "cmd": "subscribe",
            "params": {
                "channels": ["orderbook_delta"],
                "market_tickers": ["KXBTCD-26MAR31-T55000"]
            }
        }))

        # Subscribe to your own order fills
        await ws.send(json.dumps({
            "id": 2,
            "cmd": "subscribe",
            "params": {
                "channels": ["fill"]
            }
        }))

        print("Connected. Listening for updates...")

        # Listen for messages
        async for message in ws:
            data = json.loads(message)
            msg_type = data.get("type")

            if msg_type == "orderbook_delta":
                print(f"Orderbook update: {data['msg']}")
            elif msg_type == "fill":
                print(f"Fill: {data['msg']}")
            elif msg_type == "trade":
                trade = data["msg"]
                print(f"Trade: {trade['count']}x @ ${trade['yes_price']/100:.2f}")
            elif msg_type == "subscribed":
                print(f"Subscribed to: {data}")
            else:
                print(f"Message: {data}")


# Run the WebSocket client
asyncio.run(connect_kalshi_ws(
    key_id="your-api-key-id",
    private_key_path="kalshi_private_key.pem"
))

Keeping the Connection Alive

WebSocket connections may time out if idle. Send periodic ping messages to keep the connection alive:

async def heartbeat(ws, interval=30):
    """Send periodic pings to keep WebSocket alive."""
    while True:
        try:
            await ws.ping()
            await asyncio.sleep(interval)
        except websockets.exceptions.ConnectionClosed:
            break


# Use with asyncio.gather in your main loop:
# await asyncio.gather(listen_messages(ws), heartbeat(ws))

Kalshi API Python SDK

Kalshi provides an official Python SDK (kalshi-python) that wraps the REST API with convenient methods. This is the easiest way to get started with the Kalshi API Python integration:

# Install the official SDK
pip install kalshi-python

SDK Quick Start

import kalshi_python
from kalshi_python.models import *

# Configure the SDK
config = kalshi_python.Configuration()
config.host = "https://api.elections.kalshi.com/trade-api/v2"

# Create API client with your credentials
kalshi_api = kalshi_python.ApiInstance(
    email="[email protected]",
    password="your-password",
    configuration=config,
)

# --- Market Data ---
# Get all open events
events = kalshi_api.get_events(status="open", limit=20)
for event in events.events:
    print(f"{event.event_ticker}: {event.title}")

# Get a specific market
market = kalshi_api.get_market("KXBTCD-26MAR31-T55000")
print(f"Yes price: ${market.market.yes_bid / 100:.2f}")

# Get the orderbook
orderbook = kalshi_api.get_market_orderbook("KXBTCD-26MAR31-T55000")

# --- Trading ---
# Place a limit order
order_params = CreateOrderRequest(
    ticker="KXBTCD-26MAR31-T55000",
    action="buy",
    side="yes",
    type="limit",
    count=10,
    yes_price=45,
)
order = kalshi_api.create_order(order_params)
print(f"Order placed: {order.order.order_id}")

# Get your positions
positions = kalshi_api.get_positions(status="open")
for pos in positions.market_positions:
    print(f"{pos.ticker}: {pos.position} contracts")

# Get balance
balance = kalshi_api.get_balance()
print(f"Balance: ${balance.balance / 100:.2f}")

For other languages, the Kalshi REST API follows standard conventions and works with any HTTP client. See the prediction market API comparison for multi-platform integration approaches.

Rate Limits & Best Practices

The Kalshi API enforces rate limits to ensure fair access and platform stability. Understanding these limits is critical for building reliable trading bots and data pipelines.

Current Rate Limits

When you exceed a rate limit, the API returns a 429 Too Many Requests response. The response headers include Retry-After indicating how long to wait.

Rate Limit Handler

import time
import requests


class RateLimitedClient(KalshiClient):
    """Kalshi API client with automatic rate limiting."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._last_request_time = 0
        self._min_interval = 0.1  # 100ms between requests (10/sec)

    def _throttle(self):
        """Enforce minimum interval between requests."""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._min_interval:
            time.sleep(self._min_interval - elapsed)
        self._last_request_time = time.time()

    def _request_with_retry(self, method: str, path: str, **kwargs) -> dict:
        """Make a request with automatic retry on rate limit."""
        max_retries = 5
        base_delay = 1.0

        for attempt in range(max_retries):
            self._throttle()

            url = self.base_url + path
            headers = self._get_auth_headers(method.upper(), path)
            response = self.session.request(
                method, url, headers=headers, **kwargs
            )

            if response.status_code == 429:
                # Rate limited - exponential backoff
                retry_after = float(
                    response.headers.get("Retry-After", base_delay)
                )
                wait_time = retry_after * (2 ** attempt)
                print(f"Rate limited. Retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
                continue

            response.raise_for_status()
            return response.json()

        raise Exception(f"Max retries exceeded for {method} {path}")

    def get(self, path: str, params: dict = None) -> dict:
        return self._request_with_retry("GET", path, params=params)

    def post(self, path: str, data: dict = None) -> dict:
        return self._request_with_retry("POST", path, json=data)

Caching Strategies

Reduce API calls by caching responses that do not change frequently:

from functools import lru_cache
import time


class CachedKalshiClient(RateLimitedClient):
    """Client with response caching for market data."""

    def __init__(self, *args, cache_ttl=60, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache_ttl = cache_ttl
        self._cache = {}

    def get_cached(self, path: str, params: dict = None) -> dict:
        """GET with time-based caching."""
        cache_key = f"{path}:{json.dumps(params, sort_keys=True)}"
        now = time.time()

        if cache_key in self._cache:
            data, timestamp = self._cache[cache_key]
            if now - timestamp < self.cache_ttl:
                return data

        data = self.get(path, params)
        self._cache[cache_key] = (data, now)
        return data

    def get_market_cached(self, ticker: str) -> dict:
        """Get market details with caching."""
        return self.get_cached(f"/markets/{ticker}")

    def invalidate_cache(self):
        """Clear all cached data."""
        self._cache.clear()

Error Handling & Status Codes

The Kalshi API uses standard HTTP status codes and returns structured JSON error responses. Proper error handling is essential for building reliable trading systems.

HTTP Status Codes

Error Response Format

# Example error response (400 Bad Request)
{
    "code": "invalid_request",
    "message": "Invalid ticker: INVALID-TICKER does not exist",
    "service_code": "exchange"
}

# Example error response (422 Insufficient Balance)
{
    "code": "insufficient_balance",
    "message": "Insufficient balance to place this order. Required: $52.00, Available: $30.00",
    "service_code": "exchange"
}

Robust Error Handler

class KalshiAPIError(Exception):
    """Custom exception for Kalshi API errors."""
    def __init__(self, status_code, code, message):
        self.status_code = status_code
        self.code = code
        self.message = message
        super().__init__(f"[{status_code}] {code}: {message}")


def handle_response(response: requests.Response) -> dict:
    """Parse API response and raise appropriate errors."""
    if response.status_code in (200, 201):
        return response.json()

    # Parse error body
    try:
        error_data = response.json()
        code = error_data.get("code", "unknown_error")
        message = error_data.get("message", "No details provided")
    except ValueError:
        code = "parse_error"
        message = response.text

    if response.status_code == 401:
        raise KalshiAPIError(401, code, f"Authentication failed: {message}")
    elif response.status_code == 422:
        raise KalshiAPIError(422, code, f"Order rejected: {message}")
    elif response.status_code == 429:
        retry_after = response.headers.get("Retry-After", "unknown")
        raise KalshiAPIError(429, code, f"Rate limited. Retry after: {retry_after}s")
    else:
        raise KalshiAPIError(response.status_code, code, message)


# Usage in trading code
try:
    order = client.post("/portfolio/orders", data={
        "ticker": "KXBTCD-26MAR31-T55000",
        "action": "buy",
        "side": "yes",
        "type": "limit",
        "count": 10,
        "yes_price": 45,
    })
    print(f"Order placed: {order['order']['order_id']}")
except KalshiAPIError as e:
    if e.code == "insufficient_balance":
        print(f"Not enough funds: {e.message}")
    elif e.code == "market_closed":
        print(f"Market is closed: {e.message}")
    else:
        print(f"API error: {e}")

Building a Kalshi Trading Bot

Now that you understand the Kalshi API endpoints, authentication, and error handling, let us put it all together into a functional trading bot. This example implements a simple mean-reversion strategy that buys when prices dip below a threshold and sells when they recover.

Complete Trading Bot Example

"""
Kalshi Trading Bot: Mean Reversion Strategy
Buys when price drops below moving average, sells when it recovers.
For educational purposes only.
"""

import time
import json
import logging
from collections import deque
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("kalshi_bot")


class MeanReversionBot:
    """Simple mean reversion trading bot for Kalshi."""

    def __init__(self, client, ticker: str, config: dict = None):
        self.client = client
        self.ticker = ticker
        self.config = config or {}

        # Strategy parameters
        self.window_size = self.config.get("window_size", 20)
        self.entry_threshold = self.config.get("entry_threshold", 0.05)
        self.exit_threshold = self.config.get("exit_threshold", 0.02)
        self.max_position = self.config.get("max_position", 50)
        self.order_size = self.config.get("order_size", 10)

        # State
        self.price_history = deque(maxlen=self.window_size)
        self.position = 0
        self.avg_entry_price = 0
        self.total_pnl = 0

    def get_current_price(self) -> float:
        """Fetch the current mid-price for the market."""
        market = self.client.get(f"/markets/{self.ticker}")
        m = market["market"]
        bid = m.get("yes_bid", 0) / 100
        ask = m.get("yes_ask", 0) / 100
        if bid > 0 and ask > 0:
            return (bid + ask) / 2
        return m.get("last_price", 50) / 100

    def get_moving_average(self) -> float:
        """Calculate the simple moving average of recent prices."""
        if len(self.price_history) < self.window_size:
            return None
        return sum(self.price_history) / len(self.price_history)

    def should_buy(self, price: float, ma: float) -> bool:
        """Determine if we should buy (price below MA by threshold)."""
        if self.position >= self.max_position:
            return False
        return price < ma * (1 - self.entry_threshold)

    def should_sell(self, price: float, ma: float) -> bool:
        """Determine if we should sell (price recovered to MA)."""
        if self.position <= 0:
            return False
        return price > ma * (1 - self.exit_threshold)

    def place_buy_order(self, price: float):
        """Place a limit buy order."""
        price_cents = int(price * 100)
        try:
            order = self.client.post("/portfolio/orders", data={
                "ticker": self.ticker,
                "action": "buy",
                "side": "yes",
                "type": "limit",
                "count": self.order_size,
                "yes_price": price_cents,
            })
            order_id = order["order"]["order_id"]
            logger.info(
                f"BUY order placed: {self.order_size} contracts "
                f"@ ${price:.2f} (ID: {order_id})"
            )
            self.position += self.order_size
            self.avg_entry_price = (
                (self.avg_entry_price * (self.position - self.order_size)
                 + price * self.order_size) / self.position
            )
        except Exception as e:
            logger.error(f"Failed to place buy order: {e}")

    def place_sell_order(self, price: float):
        """Place a limit sell order."""
        price_cents = int(price * 100)
        sell_count = min(self.order_size, self.position)
        try:
            order = self.client.post("/portfolio/orders", data={
                "ticker": self.ticker,
                "action": "sell",
                "side": "yes",
                "type": "limit",
                "count": sell_count,
                "yes_price": price_cents,
            })
            order_id = order["order"]["order_id"]
            pnl = (price - self.avg_entry_price) * sell_count
            self.total_pnl += pnl
            self.position -= sell_count
            logger.info(
                f"SELL order placed: {sell_count} contracts "
                f"@ ${price:.2f} (P&L: ${pnl:.2f}, "
                f"Total P&L: ${self.total_pnl:.2f})"
            )
        except Exception as e:
            logger.error(f"Failed to place sell order: {e}")

    def run(self, poll_interval: int = 30):
        """Main bot loop."""
        logger.info(f"Starting bot for {self.ticker}")
        logger.info(f"Config: window={self.window_size}, "
                     f"entry={self.entry_threshold}, "
                     f"exit={self.exit_threshold}")

        while True:
            try:
                # Get current price
                price = self.get_current_price()
                self.price_history.append(price)

                # Calculate moving average
                ma = self.get_moving_average()
                if ma is None:
                    logger.info(
                        f"Collecting data: {len(self.price_history)}"
                        f"/{self.window_size} prices. "
                        f"Current: ${price:.2f}"
                    )
                    time.sleep(poll_interval)
                    continue

                deviation = (price - ma) / ma * 100
                logger.info(
                    f"Price: ${price:.2f} | MA: ${ma:.2f} | "
                    f"Dev: {deviation:+.1f}% | "
                    f"Pos: {self.position} | P&L: ${self.total_pnl:.2f}"
                )

                # Check signals
                if self.should_buy(price, ma):
                    logger.info("Signal: BUY (price below MA threshold)")
                    self.place_buy_order(price)
                elif self.should_sell(price, ma):
                    logger.info("Signal: SELL (price recovered to MA)")
                    self.place_sell_order(price)

                time.sleep(poll_interval)

            except KeyboardInterrupt:
                logger.info("Bot stopped by user.")
                break
            except Exception as e:
                logger.error(f"Error in main loop: {e}")
                time.sleep(poll_interval * 2)


# --- Run the bot ---
if __name__ == "__main__":
    # Initialize API client (use demo for testing!)
    client = KalshiClient(
        key_id="your-api-key-id",
        private_key_path="kalshi_private_key.pem",
        base_url="https://demo-api.kalshi.co/trade-api/v2"  # DEMO
    )

    bot = MeanReversionBot(
        client=client,
        ticker="KXBTCD-26MAR31-T55000",
        config={
            "window_size": 20,        # lookback window for MA
            "entry_threshold": 0.05,  # buy when 5% below MA
            "exit_threshold": 0.02,   # sell when within 2% of MA
            "max_position": 50,       # max contracts to hold
            "order_size": 10,         # contracts per order
        }
    )

    bot.run(poll_interval=30)  # check every 30 seconds

Bot Architecture Recommendations

For production trading bots on the Kalshi trading API, consider these improvements:

• Use websockets instead of polling. The REST polling approach above is simple but adds latency. For competitive trading, switch to the websocket API for real-time price updates.
• Implement position tracking from fills. Track your actual position by monitoring fill events rather than assuming orders execute immediately at the placed price.
• Add risk management. Implement stop-loss logic, maximum drawdown limits, and position size limits relative to your account balance.
• Handle exchange closures. Check exchange status before trading and gracefully handle maintenance windows.
• Log everything. Store all orders, fills, prices, and decisions for post-trade analysis.
• Run on a VPS or cloud server. For consistent connectivity, run your bot on a cloud server (AWS, DigitalOcean, etc.) rather than your local machine.

Kalshi API vs Polymarket API

If you are deciding between integrating with the Kalshi API or the Polymarket API, here is how they compare across key dimensions:

For a comprehensive overview of all prediction market APIs, see our prediction market API comparison guide.

Frequently Asked Questions

Related Guides

• Polymarket API Guide — Complete guide to the Polymarket CLOB API with Python examples
• How to Build a Polymarket Bot — Step-by-step guide to automated prediction market trading
• Kalshi Fees Explained — Complete 2026 breakdown of trading, deposit, and withdrawal costs
• How Does Kalshi Work? — Beginner's guide to the CFTC-regulated prediction market
• Prediction Market Strategies — Proven approaches to profitable event trading
• Kalshi App Review — Mobile and desktop app features for prediction market trading
• Prediction Market API Comparison — Compare APIs across Kalshi, Polymarket, and more
• Prediction Market Arbitrage — Find risk-free profit opportunities across platforms
• What Are Prediction Markets? — Complete introduction to event trading
• Best Prediction Markets 2026 — Top platforms ranked by fees, features, and markets