Kalshi API: Complete Trading Automation Guide (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://trading-api.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.MAX_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://trading-api.kalshi.com/trade-api/v2"
)

import kalshi_python

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

kalshi_api = kalshi_python.ApiInstance(
    email="[email protected]",
    password="your-password",
    configuration=config
)

# Check exchange status
status = kalshi_api.get_exchange_status()
print(status)

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

for market in response.json()["markets"]:
    print(f"{market['ticker']}: {market['title']}")
    print(f"  Yes price: {market.get('yes_ask', '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"]

print("YES bids:", book.get("yes", []))
print("NO bids:", book.get("no", []))

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"
    "type": "limit",         # "limit" or "market"
    "action": "buy",         # "buy" or "sell"
    "count": 10,             # Number of contracts
    "yes_price": 65,         # Price in cents (65 = $0.65)
    "client_order_id": str(uuid.uuid4())  # Your tracking ID
}

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

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

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

response = client.post(f"/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("/markets/history", params={
    "ticker": "KXNFLGAME-25OCT12CLEPIT"
})

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.MAX_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())

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"),
            "yes_ask": market.get("yes_ask"),
            "no_bid": market.get("no_bid"),
            "no_ask": market.get("no_ask"),
            "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: int):
        """Place a limit order"""
        order = {
            "ticker": self.ticker,
            "side": side,
            "type": "limit",
            "action": action,
            "count": count,
            "yes_price": price,
            "client_order_id": str(uuid.uuid4())
        }
        response = self.client.post("/orders", data=order)

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

        return response

    def run(self, buy_threshold: int, sell_threshold: int, interval: int = 10):
        """
        Simple threshold strategy:
        - Buy YES if price drops below buy_threshold
        - Sell YES if price rises above sell_threshold
        """
        logger.info(f"Starting bot for {self.ticker}")
        logger.info(f"Buy below: {buy_threshold}c | Sell above: {sell_threshold}c")

        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}c ask={yes_ask}c | "
                    f"Position: {self.position} contracts"
                )

                # Buy signal
                if (yes_ask and yes_ask <= buy_threshold
                        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 yes_bid >= sell_threshold
                        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=40, sell_threshold=60, interval=10)

def scan_markets(client, min_volume=1000, max_yes_price=30):
    """Find underpriced YES contracts with decent volume"""
    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")

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

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

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 = kalshi_market.get("yes_ask", 100) / 100
            poly_no = polymarket_markets[ticker]["no_price"]

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

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 ask price for our side
            price_key = f"{side}_ask"
            price = market.get(price_key)

            if price:
                client.post("/orders", data={
                    "ticker": ticker,
                    "side": side,
                    "type": "limit",
                    "action": "buy",
                    "count": 5,
                    "yes_price": price if side == "yes" else (100 - 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 = pos.get("market_exposure", 0) / 100
        total_exposure += exposure

        print(f"  {pos['ticker']}")
        print(f"    Side: {pos.get('side', 'N/A')}")
        print(f"    Contracts: {pos.get('position', 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}")