Prediction Markets: Polymarket & Kalshi

pip install swarms py-clob-client requests cryptography

import requests
from typing import Optional


GAMMA_URL = "https://gamma-api.polymarket.com"
CLOB_URL = "https://clob.polymarket.com"


def discover_polymarket_events(
    limit: int = 10,
    tag: Optional[str] = None,
) -> str:
    """
    Discover active prediction markets on Polymarket.

    Args:
        limit: Number of events to return (max 100).
        tag: Optional category filter (e.g., 'politics', 'crypto', 'sports').

    Returns:
        str: Formatted string of active markets with questions, odds, and volume.
    """
    params = {
        "active": "true",
        "closed": "false",
        "order": "volume_24hr",
        "ascending": "false",
        "limit": limit,
    }
    if tag:
        params["tag"] = tag

    resp = requests.get(f"{GAMMA_URL}/events", params=params, timeout=30)
    resp.raise_for_status()
    events = resp.json()

    results = []
    for event in events:
        results.append(f"Event: {event['title']}")
        for market in event.get("markets", []):
            prices = market.get("outcomePrices", ["?", "?"])
            token_ids = market.get("clobTokenIds", [])
            results.append(
                f"  Market: {market['question']}\n"
                f"  Outcomes: {market.get('outcomes', [])}\n"
                f"  Prices: Yes={prices[0]}, No={prices[1]}\n"
                f"  24h Volume: ${market.get('volume24hr', 0):,.0f}\n"
                f"  Token IDs: {token_ids}"
            )
    return "\n".join(results) if results else "No active markets found."


def get_polymarket_orderbook(token_id: str) -> str:
    """
    Get the current order book for a Polymarket token.

    Args:
        token_id: The CLOB token ID for the market outcome.

    Returns:
        str: Order book summary with best bid, ask, midpoint, and spread.
    """
    mid = requests.get(
        f"{CLOB_URL}/midpoint", params={"token_id": token_id}, timeout=10
    ).json()
    spread = requests.get(
        f"{CLOB_URL}/spread", params={"token_id": token_id}, timeout=10
    ).json()
    book = requests.get(
        f"{CLOB_URL}/book", params={"token_id": token_id}, timeout=10
    ).json()

    top_bids = book.get("bids", [])[:3]
    top_asks = book.get("asks", [])[:3]

    return (
        f"Midpoint: {mid.get('mid', 'N/A')}\n"
        f"Spread: {spread.get('spread', 'N/A')}\n"
        f"Top 3 Bids: {top_bids}\n"
        f"Top 3 Asks: {top_asks}"
    )

import datetime
import base64
from urllib.parse import urlparse
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

PREDICTION_MARKET_PROMPT = """You are an expert prediction market analyst.

Your workflow:
1. Use your tools to discover active markets on Polymarket and 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="Prediction-Market-Analyst",
    agent_description="Analyzes prediction markets and identifies edges",
    system_prompt=PREDICTION_MARKET_PROMPT,
    model_name="gpt-4o",
    max_loops=3,
    tools=[
        discover_polymarket_events,
        discover_kalshi_events,
        get_polymarket_orderbook,
        get_kalshi_orderbook,
    ],
    output_type="str",
)

result = analyst.run(
    "Find the top 5 most active prediction markets across Polymarket "
    "and 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 json
from typing import Optional

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


# --- Polymarket Execution ---

def place_polymarket_order(
    token_id: str,
    side: str,
    price: float,
    size: float,
) -> str:
    """
    Place a limit order on Polymarket.

    Args:
        token_id: The CLOB token ID for the outcome.
        side: 'BUY' or 'SELL'.
        price: Limit price between 0.01 and 0.99.
        size: Number of shares to buy/sell.

    Returns:
        str: Order confirmation or dry-run summary.
    """
    if DRY_RUN:
        return (
            f"[DRY RUN] Polymarket order: {side} {size} shares "
            f"at ${price:.2f} for token {token_id[:20]}..."
        )

    from py_clob_client.client import ClobClient
    from py_clob_client.clob_types import OrderArgs, ApiCreds
    from py_clob_client.order_builder.constants import BUY, SELL

    client = ClobClient(
        host="https://clob.polymarket.com",
        key=os.environ["POLYMARKET_PRIVATE_KEY"],
        chain=137,
        creds=ApiCreds(
            api_key=os.environ["POLY_API_KEY"],
            api_secret=os.environ["POLY_API_SECRET"],
            api_passphrase=os.environ["POLY_API_PASSPHRASE"],
        ),
        funder=os.environ["POLYMARKET_FUNDER_ADDRESS"],
    )

    order_side = BUY if side.upper() == "BUY" else SELL
    resp = client.create_and_post_order(
        OrderArgs(
            token_id=token_id,
            price=price,
            size=size,
            side=order_side,
        ),
    )
    return f"Order placed: ID={resp['orderID']}, Status={resp['status']}"


# --- Kalshi Execution ---

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}"
        )

    import uuid

    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="Market-Researcher",
    agent_description="Discovers and summarizes prediction markets",
    system_prompt="""You are a prediction market researcher. Your job:
1. Use your tools to discover active markets on both Polymarket and 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_polymarket_events,
        discover_kalshi_events,
        get_polymarket_orderbook,
        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 prediction 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 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 prediction market 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:
   - Platform (Polymarket or Kalshi)
   - Token ID or ticker
   - Side (BUY/SELL, YES/NO)
   - Size (number of contracts/shares)
   - Limit price
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 Polymarket and Kalshi",
    system_prompt="""You are a trade execution agent. Given approved trades from the risk manager:

1. Parse each approved trade's parameters
2. Use the appropriate tool to place the order (Polymarket or Kalshi)
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_polymarket_order, place_kalshi_order],
    output_type="str",
)

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

result = swarm.run(
    "Scan Polymarket and 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(
    "trades.log",
    rotation="1 day",
    format="{time} | {level} | {message}",
)

# In your execution tool, log every order:
logger.info(json.dumps({
    "action": "order_placed",
    "platform": "polymarket",
    "token_id": token_id,
    "side": side,
    "price": price,
    "size": size,
    "dry_run": DRY_RUN,
}))