Prediction Markets: Kalshi

pip install swarms requests cryptography

import requests
import datetime
import base64
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import padding


KALSHI_BASE = "https://api.elections.kalshi.com/trade-api/v2"
KALSHI_DEMO_BASE = "https://demo-api.kalshi.co/trade-api/v2"


def discover_kalshi_events(limit: int = 10) -> str:
    """
    Discover active prediction markets on Kalshi.

    Args:
        limit: Number of events to return (max 200).

    Returns:
        str: Formatted string of active markets with tickers, prices, and volume.
    """
    resp = requests.get(
        f"{KALSHI_BASE}/events",
        params={
            "status": "open",
            "limit": limit,
            "with_nested_markets": True,
        },
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()

    results = []
    for event in data.get("events", []):
        results.append(f"Event: {event['title']} ({event['event_ticker']})")
        for m in event.get("markets", []):
            results.append(
                f"  Market: {m['ticker']}\n"
                f"  Yes Bid: ${m.get('yes_bid_dollars', '?')} | "
                f"Yes Ask: ${m.get('yes_ask_dollars', '?')}\n"
                f"  Last Price: ${m.get('last_price_dollars', '?')}\n"
                f"  Volume 24h: {m.get('volume_24h_fp', '?')}\n"
                f"  Closes: {m.get('close_time', '?')}"
            )
    return "\n".join(results) if results else "No active events found."


def get_kalshi_orderbook(ticker: str) -> str:
    """
    Get the current order book for a Kalshi market.

    Args:
        ticker: The market ticker (e.g., 'KXHIGHNY-25APR27-T55').

    Returns:
        str: Order book summary with YES/NO bids and implied spread.
    """
    resp = requests.get(
        f"{KALSHI_BASE}/markets/{ticker}/orderbook",
        params={"depth": 5},
        timeout=10,
    )
    resp.raise_for_status()
    book = resp.json().get("orderbook_fp", {})

    yes_bids = book.get("yes_dollars", [])
    no_bids = book.get("no_dollars", [])

    summary = f"Ticker: {ticker}\n"
    if yes_bids:
        best_yes_bid = float(yes_bids[-1][0])
        summary += f"Best YES Bid: ${best_yes_bid:.4f}\n"
    if no_bids:
        best_no_bid = float(no_bids[-1][0])
        best_yes_ask = 1.0 - best_no_bid
        summary += f"Best YES Ask: ${best_yes_ask:.4f}\n"
    if yes_bids and no_bids:
        spread = best_yes_ask - best_yes_bid
        summary += f"Spread: ${spread:.4f}\n"
    summary += f"YES depth: {len(yes_bids)} levels | NO depth: {len(no_bids)} levels"
    return summary

from swarms import Agent

KALSHI_ANALYST_PROMPT = """You are an expert Kalshi prediction market analyst.

Your workflow:
1. Use your tools to discover active markets on Kalshi
2. Select the most interesting markets with high volume and liquidity
3. For each selected market, research the event and estimate the TRUE probability
4. Compare your estimated probability to the market's implied probability
5. Identify edges: markets where your estimate differs from market odds by >10%

When analyzing a market:
- State the question clearly
- List key factors that influence the outcome
- Estimate the probability with reasoning
- Compare to market price
- Recommend: BET YES, BET NO, or NO EDGE

Always express probabilities as percentages and explain your reasoning.
Never recommend betting more than 5% of bankroll on a single position.
"""

analyst = Agent(
    agent_name="Kalshi-Analyst",
    agent_description="Analyzes Kalshi prediction markets and identifies edges",
    system_prompt=KALSHI_ANALYST_PROMPT,
    model_name="gpt-4o",
    max_loops=3,
    tools=[
        discover_kalshi_events,
        get_kalshi_orderbook,
    ],
    output_type="str",
)

result = analyst.run(
    "Find the top 5 most active prediction markets on Kalshi. "
    "Analyze each one and identify any markets where you believe there is "
    "a significant edge (>10% probability difference between your estimate "
    "and the market price)."
)
print(result)

import os
import uuid


# Dry-run mode — set to False only when ready for live trading
DRY_RUN = True


def _kalshi_auth_headers(method: str, path: str) -> dict:
    """Build Kalshi authentication headers using RSA-PSS signing."""
    key_path = os.environ["KALSHI_PRIVATE_KEY_PATH"]
    api_key_id = os.environ["KALSHI_API_KEY_ID"]

    with open(key_path, "rb") as f:
        private_key = serialization.load_pem_private_key(
            f.read(), password=None, backend=default_backend()
        )

    timestamp = str(int(datetime.datetime.now().timestamp() * 1000))
    path_clean = path.split("?")[0]
    message = f"{timestamp}{method}{path_clean}".encode("utf-8")
    signature = private_key.sign(
        message,
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.DIGEST_LENGTH,
        ),
        hashes.SHA256(),
    )

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


def place_kalshi_order(
    ticker: str,
    side: str,
    action: str,
    count: int,
    price_cents: int,
) -> str:
    """
    Place a limit order on Kalshi.

    Args:
        ticker: Market ticker (e.g., 'KXHIGHNY-25APR27-T55').
        side: 'yes' or 'no'.
        action: 'buy' or 'sell'.
        count: Number of contracts.
        price_cents: Limit price in cents (1-99).

    Returns:
        str: Order confirmation or dry-run summary.
    """
    if DRY_RUN:
        return (
            f"[DRY RUN] Kalshi order: {action} {count} {side} contracts "
            f"at {price_cents}c on {ticker}"
        )

    path = "/trade-api/v2/portfolio/orders"
    headers = _kalshi_auth_headers("POST", path)
    body = {
        "ticker": ticker,
        "action": action,
        "side": side,
        "type": "limit",
        "count": count,
        "yes_price": price_cents if side == "yes" else None,
        "no_price": price_cents if side == "no" else None,
        "client_order_id": str(uuid.uuid4()),
    }
    body = {k: v for k, v in body.items() if v is not None}

    resp = requests.post(
        KALSHI_BASE + "/portfolio/orders", headers=headers, json=body, timeout=30
    )
    resp.raise_for_status()
    order = resp.json().get("order", {})
    return f"Order placed: ID={order.get('order_id')}, Status={order.get('status')}"

from swarms import Agent, SequentialWorkflow

# --- Research Agent ---
research_agent = Agent(
    agent_name="Kalshi-Researcher",
    agent_description="Discovers and summarizes Kalshi markets",
    system_prompt="""You are a Kalshi researcher. Your job:
1. Use your tools to discover active markets on Kalshi
2. Focus on markets with high volume and liquidity
3. For each market, provide: the question, current odds, volume, and closing date
4. Gather relevant context about each event from your knowledge
5. Output a structured research brief for each market

Format your output as a clear research report that an analyst can use.""",
    model_name="gpt-4o",
    max_loops=2,
    tools=[
        discover_kalshi_events,
        get_kalshi_orderbook,
    ],
    output_type="str",
)

# --- Analyst Agent ---
analyst_agent = Agent(
    agent_name="Probability-Analyst",
    agent_description="Estimates true probabilities and finds edges",
    system_prompt="""You are a quantitative analyst specializing in probability estimation.
Given a research brief on Kalshi markets, you must:

1. For each market, estimate the TRUE probability of each outcome
2. Show your reasoning: list base rates, key factors, and analogies
3. Compare your estimate to the market's implied probability
4. Calculate the edge: (your estimate - market price) / market price
5. Flag any market where the absolute edge exceeds 10%

Output a structured analysis with:
- Market ticker and question
- Your probability estimate (with confidence interval)
- Market implied probability
- Edge percentage
- Verdict: STRONG BUY YES, LEAN BUY YES, NO EDGE, LEAN BUY NO, STRONG BUY NO""",
    model_name="gpt-4o",
    max_loops=1,
    output_type="str",
)

# --- Risk Agent ---
risk_agent = Agent(
    agent_name="Risk-Manager",
    agent_description="Enforces position sizing and risk limits",
    system_prompt="""You are a risk manager for a Kalshi trading operation.
Given the analyst's recommendations, you must:

1. Filter out any recommendations with edge < 10%
2. Apply Kelly Criterion for position sizing (use half-Kelly for safety)
3. Enforce these hard limits:
   - Max 5% of bankroll on any single position
   - Max 20% of bankroll in correlated positions
   - No positions in markets closing within 1 hour (too volatile)
   - Minimum liquidity: $10,000 in 24h volume
4. For approved trades, output the exact order parameters:
   - Ticker
   - Side (yes/no)
   - Action (buy/sell)
   - Count (number of contracts)
   - Limit price in cents (1-99)
5. For rejected trades, explain why

Assume a bankroll of $1,000 unless specified otherwise.""",
    model_name="gpt-4o",
    max_loops=1,
    output_type="str",
)

# --- Execution Agent ---
execution_agent = Agent(
    agent_name="Trade-Executor",
    agent_description="Executes approved trades on Kalshi",
    system_prompt="""You are a Kalshi trade execution agent. Given approved trades from the risk manager:

1. Parse each approved trade's parameters
2. Use the place_kalshi_order tool to place the order
3. Report the result of each order (confirmation or error)
4. If an order fails, do NOT retry — report the failure

Always log every action. Never modify the risk manager's parameters.""",
    model_name="gpt-4o",
    max_loops=1,
    tools=[place_kalshi_order],
    output_type="str",
)

swarm = SequentialWorkflow(
    agents=[research_agent, analyst_agent, risk_agent, execution_agent],
    max_loops=1,
)

result = swarm.run(
    "Scan Kalshi for the top active markets. "
    "Identify any edges and execute approved trades. "
    "Bankroll: $1,000."
)
print(result)

MAX_POSITION_PCT = 0.05      # 5% of bankroll per position
MAX_CORRELATED_PCT = 0.20    # 20% in correlated markets
MIN_VOLUME_24H = 10_000      # $10k minimum 24h volume
MIN_EDGE_PCT = 0.10          # 10% minimum edge to trade

from loguru import logger
import json

logger.add(
    "kalshi-trades.log",
    rotation="1 day",
    format="{time} | {level} | {message}",
)

# In your execution tool, log every order:
logger.info(json.dumps({
    "action": "order_placed",
    "platform": "kalshi",
    "ticker": ticker,
    "side": side,
    "action_type": action,
    "count": count,
    "price_cents": price_cents,
    "dry_run": DRY_RUN,
}))