Kalshi API Guide: Python SDK Setup, RSA Auth & Demo Sandbox (2026)

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

class KalshiClient:
    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"

        with open(private_key_path, "rb") as f:
            self.private_key = serialization.load_pem_private_key(
                f.read(), password=None
            )

    def _sign(self, timestamp_ms: str, method: str, path: str) -> str:
        """Sign the request using RSA-PSS"""
        message = f"{timestamp_ms}{method}{path}".encode("utf-8")
        signature = self.private_key.sign(
            message,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.DIGEST_LENGTH
            ),
            hashes.SHA256()
        )
        return base64.b64encode(signature).decode("utf-8")

    def _headers(self, method: str, path: str) -> dict:
        """Generate authenticated headers for a request"""
        timestamp = str(int(time.time() * 1000))
        signature = self._sign(timestamp, method, path)
        return {
            "KALSHI-ACCESS-KEY": self.key_id,
            "KALSHI-ACCESS-SIGNATURE": signature,
            "KALSHI-ACCESS-TIMESTAMP": timestamp,
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

    def get(self, path: str, params: dict = None):
        url = f"{self.base_url}{path}"
        headers = self._headers("GET", path)
        return requests.get(url, headers=headers, params=params)

    def post(self, path: str, data: dict = None):
        url = f"{self.base_url}{path}"
        headers = self._headers("POST", path)
        return requests.post(url, headers=headers, json=data)

    def delete(self, path: str, data: dict = None):
        url = f"{self.base_url}{path}"
        headers = self._headers("DELETE", path)
        return requests.delete(url, headers=headers, json=data)

# For demo/testing:
client = KalshiClient(
    key_id="your-demo-api-key-id",
    private_key_path="path/to/demo-private-key.pem",
    base_url="https://demo-api.kalshi.co/trade-api/v2"
)

# For production:
client = KalshiClient(
    key_id="your-prod-api-key-id",
    private_key_path="path/to/prod-private-key.pem",
    base_url="https://api.elections.kalshi.com/trade-api/v2"
)

# Synchronous (recommended for most bots)
pip install kalshi_python_sync

# Asynchronous (for async/await workflows)
pip install kalshi_python_async

from kalshi_python_sync import Configuration, KalshiClient

config = Configuration(
    host="https://demo-api.kalshi.co/trade-api/v2"  # Demo
)

# Read private key from file
with open("path/to/private_key.pem", "r") as f:
    private_key = f.read()

config.api_key_id = "your-api-key-id"
config.private_key_pem = private_key

client = KalshiClient(config)

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

npm install kalshi-typescript

import { Configuration, PortfolioApi } from 'kalshi-typescript';

const config = new Configuration({
    apiKey: 'your-api-key-id',
    privateKeyPath: 'path/to/your/private-key.pem',
    basePath: 'https://api.elections.kalshi.com/trade-api/v2'
});

const portfolioApi = new PortfolioApi(config);
const balance = await portfolioApi.getBalance();

response = client.get("/markets", params={
    "limit": 20,
    "status": "open"
})

for market in response.json()["markets"]:
    print(f"{market['ticker']}: {market['title']}")
    print(f"  Yes bid: {market.get('yes_bid_dollars', 'N/A')}")
    print(f"  Volume: {market.get('volume', 0)}")

response = client.get("/markets/KXNFLGAME-25OCT12CLEPIT")
market = response.json()["market"]

response = client.get("/markets/KXNFLGAME-25OCT12CLEPIT/orderbook")
book = response.json()["orderbook_fp"]

# Each entry is [price_dollars, count_fp] — both strings
print("YES bids:", book.get("yes_dollars", []))
print("NO bids:", book.get("no_dollars", []))

# Example: [["0.4200", "13.00"], ["0.3500", "11.00"]]
# Sorted ascending — last element is the best (highest) bid

response = client.get("/events", params={"status": "open"})
events = response.json()["events"]

response = client.get("/series")

import uuid

order = {
    "ticker": "KXNFLGAME-25OCT12CLEPIT",
    "side": "yes",                        # "yes" or "no"
    "action": "buy",                      # "buy" or "sell"
    "count_fp": "10.00",                  # Contract quantity (fixed-point string)
    "yes_price_dollars": "0.6500",        # Price as dollar string
    "client_order_id": str(uuid.uuid4())  # Idempotent tracking ID
}

response = client.post("/portfolio/orders", data=order)
print(response.json())

response = client.get("/portfolio/orders", params={
    "status": "resting"  # resting, canceled, executed
})

response = client.delete(f"/portfolio/orders/{order_id}")

response = client.post(f"/portfolio/orders/{order_id}/decrease", data={
    "reduce_by": 5
})

response = client.get("/portfolio/positions")
positions = response.json()["positions"]

for pos in positions:
    print(f"{pos['ticker']}: {pos['position']} contracts")
    print(f"  Market exposure: ${pos.get('market_exposure', 0) / 100:.2f}")

response = client.get("/portfolio/balance")
balance = response.json()
print(f"Available: ${balance['balance'] / 100:.2f}")

response = client.get("/portfolio/settlements")

response = client.get("/historical/cutoff")
cutoffs = response.json()
# Returns: market_settled_ts, trades_created_ts, orders_updated_ts

# Example: fetch historical trades
response = client.get("/historical/trades", params={
    "ticker": "KXNFLGAME-25OCT12CLEPIT",
    "limit": 100
})

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

KEY_ID = "your-api-key-id"
PRIVATE_KEY_PATH = "path/to/private-key.pem"
WS_URL = "wss://demo-api.kalshi.co/trade-api/ws/v2"  # Demo

def load_private_key():
    with open(PRIVATE_KEY_PATH, "rb") as f:
        return serialization.load_pem_private_key(f.read(), password=None)

def sign_ws_message(private_key, timestamp_ms: str) -> str:
    message = f"{timestamp_ms}GET/trade-api/ws/v2".encode("utf-8")
    signature = private_key.sign(
        message,
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.DIGEST_LENGTH
        ),
        hashes.SHA256()
    )
    return base64.b64encode(signature).decode("utf-8")

async def stream_kalshi():
    private_key = load_private_key()
    timestamp = str(int(time.time() * 1000))
    signature = sign_ws_message(private_key, timestamp)

    headers = {
        "KALSHI-ACCESS-KEY": KEY_ID,
        "KALSHI-ACCESS-SIGNATURE": signature,
        "KALSHI-ACCESS-TIMESTAMP": timestamp
    }

    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        print("Connected to Kalshi WebSocket")

        # Subscribe to ticker updates
        await ws.send(json.dumps({
            "id": 1,
            "cmd": "subscribe",
            "params": {
                "channels": ["ticker"]
            }
        }))

        # Subscribe to orderbook for a specific market
        await ws.send(json.dumps({
            "id": 2,
            "cmd": "subscribe",
            "params": {
                "channels": ["orderbook_delta"],
                "market_tickers": ["KXNFLGAME-25OCT12CLEPIT"]
            }
        }))

        async for message in ws:
            data = json.loads(message)
            print(f"Update: {data}")

asyncio.run(stream_kalshi())

# Add markets to an existing subscription
await ws.send(json.dumps({
    "id": 3,
    "cmd": "update_subscription",
    "params": {
        "sids": [1],
        "market_tickers": ["NEW-MARKET-1", "NEW-MARKET-2"],
        "action": "add_markets"
    }
}))

import time
import uuid
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("kalshi-bot")

class SimpleKalshiBot:
    def __init__(self, client: KalshiClient, ticker: str):
        self.client = client
        self.ticker = ticker
        self.position = 0
        self.max_position = 50  # Max contracts to hold

    def get_market_price(self) -> dict:
        """Fetch current market prices"""
        response = self.client.get(f"/markets/{self.ticker}")
        market = response.json()["market"]
        return {
            "yes_bid": market.get("yes_bid_dollars"),
            "yes_ask": market.get("yes_ask_dollars"),
            "no_bid": market.get("no_bid_dollars"),
            "no_ask": market.get("no_ask_dollars"),
            "volume": market.get("volume", 0)
        }

    def check_position(self):
        """Get current position in this market"""
        response = self.client.get("/portfolio/positions", params={
            "ticker": self.ticker
        })
        positions = response.json().get("positions", [])
        for pos in positions:
            if pos["ticker"] == self.ticker:
                self.position = pos.get("position", 0)
                return
        self.position = 0

    def place_order(self, side: str, action: str, count: int, price_dollars: str):
        """Place a limit order"""
        order = {
            "ticker": self.ticker,
            "side": side,
            "action": action,
            "count_fp": f"{count}.00",
            "yes_price_dollars": price_dollars,
            "client_order_id": str(uuid.uuid4())
        }
        response = self.client.post("/portfolio/orders", data=order)

        if response.status_code == 201:
            logger.info(f"Order placed: {side} {action} {count}x @ ${price_dollars}")
        else:
            logger.error(f"Order failed: {response.text}")

        return response

    def run(self, buy_threshold: str, sell_threshold: str, interval: int = 10):
        """
        Simple threshold strategy:
        - Buy YES if price drops below buy_threshold
        - Sell YES if price rises above sell_threshold
        Thresholds are dollar strings (e.g., "0.40", "0.60")
        """
        from decimal import Decimal
        buy_t = Decimal(buy_threshold)
        sell_t = Decimal(sell_threshold)

        logger.info(f"Starting bot for {self.ticker}")
        logger.info(f"Buy below: ${buy_threshold} | Sell above: ${sell_threshold}")

        while True:
            try:
                prices = self.get_market_price()
                self.check_position()

                yes_ask = prices["yes_ask"]
                yes_bid = prices["yes_bid"]

                logger.info(
                    f"Price: bid=${yes_bid} ask=${yes_ask} | "
                    f"Position: {self.position} contracts"
                )

                # Buy signal
                if (yes_ask and Decimal(yes_ask) <= buy_t
                        and self.position < self.max_position):
                    contracts = min(5, self.max_position - self.position)
                    self.place_order("yes", "buy", contracts, yes_ask)

                # Sell signal
                if (yes_bid and Decimal(yes_bid) >= sell_t
                        and self.position > 0):
                    contracts = min(5, self.position)
                    self.place_order("yes", "sell", contracts, yes_bid)

            except Exception as e:
                logger.error(f"Error: {e}")

            time.sleep(interval)


# Usage:
client = KalshiClient(
    key_id="your-key-id",
    private_key_path="path/to/key.pem",
    base_url="https://demo-api.kalshi.co/trade-api/v2"  # START WITH DEMO
)

bot = SimpleKalshiBot(client, ticker="KXNFLGAME-25OCT12CLEPIT")
bot.run(buy_threshold="0.40", sell_threshold="0.60", interval=10)

from decimal import Decimal

def scan_markets(client, min_volume=1000, max_yes_price="0.30"):
    """Find underpriced YES contracts with decent volume"""
    max_price = Decimal(max_yes_price)
    response = client.get("/markets", params={
        "status": "open",
        "limit": 200
    })

    opportunities = []
    for market in response.json()["markets"]:
        vol = market.get("volume", 0)
        yes_ask = market.get("yes_ask_dollars")

        if vol >= min_volume and yes_ask and Decimal(yes_ask) <= max_price:
            opportunities.append({
                "ticker": market["ticker"],
                "title": market["title"],
                "yes_ask_dollars": yes_ask,
                "volume": vol
            })

    return sorted(opportunities, key=lambda x: x["volume"], reverse=True)

from decimal import Decimal

def find_cross_platform_arbs(kalshi_client, polymarket_markets):
    """
    Compare Kalshi and Polymarket prices for matching events.
    Requires a mapping of Kalshi tickers to Polymarket token IDs.
    """
    kalshi_response = kalshi_client.get("/markets", params={
        "status": "open", "limit": 200
    })

    for kalshi_market in kalshi_response.json()["markets"]:
        ticker = kalshi_market["ticker"]

        # Check if we have a Polymarket match
        if ticker in polymarket_markets:
            kalshi_yes = Decimal(kalshi_market.get("yes_ask_dollars", "1.00"))
            poly_no = Decimal(str(polymarket_markets[ticker]["no_price"]))

            combined = kalshi_yes + poly_no
            if combined < Decimal("0.98"):  # 2%+ arb
                print(f"ARB: {ticker}")
                print(f"  Kalshi YES: ${kalshi_yes}")
                print(f"  Poly NO: ${poly_no}")
                print(f"  Combined: ${combined}")
                print(f"  Edge: ${Decimal('1.00') - combined}")

import time

def news_driven_strategy(client, ticker, news_monitor):
    """
    When positive news hits, buy YES.
    When negative news hits, buy NO.
    """
    while True:
        signal = news_monitor.get_latest_signal(ticker)

        if signal and signal["confidence"] > 0.7:
            side = "yes" if signal["direction"] == "positive" else "no"
            market = client.get(f"/markets/{ticker}").json()["market"]

            # Get current bid/ask price for our side
            price_key = f"{side}_ask_dollars"
            price = market.get(price_key)

            if price:
                client.post("/portfolio/orders", data={
                    "ticker": ticker,
                    "side": side,
                    "action": "buy",
                    "count_fp": "5.00",
                    "yes_price_dollars": price,
                    "client_order_id": str(uuid.uuid4())
                })

        time.sleep(30)

def portfolio_summary(client):
    """Get a complete portfolio summary"""
    balance = client.get("/portfolio/balance").json()
    positions = client.get("/portfolio/positions").json().get("positions", [])

    total_exposure = 0
    total_unrealized = 0

    print(f"Cash Balance: ${balance['balance'] / 100:.2f}")
    print(f"\nOpen Positions:")
    print("-" * 60)

    for pos in positions:
        exposure = float(pos.get("market_exposure_dollars", "0"))
        total_exposure += exposure

        print(f"  {pos['ticker']}")
        print(f"    Side: {pos.get('side', 'N/A')}")
        print(f"    Contracts: {pos.get('position_fp', '0')}")
        print(f"    Exposure: ${exposure:.2f}")

    print("-" * 60)
    print(f"Total Exposure: ${total_exposure:.2f}")
    print(f"Total Portfolio: ${(balance['balance'] / 100) + total_exposure:.2f}")