Prediction Markets: Polymarket

pip install swarms py-clob-client requests

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

from swarms import Agent

POLYMARKET_ANALYST_PROMPT = """You are an expert Polymarket prediction market analyst.

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

result = analyst.run(
    "Find the top 5 most active prediction markets on Polymarket. "
    "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

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


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']}"

from swarms import Agent, SequentialWorkflow

# --- Research Agent ---
research_agent = Agent(
    agent_name="Polymarket-Researcher",
    agent_description="Discovers and summarizes Polymarket markets",
    system_prompt="""You are a Polymarket researcher. Your job:
1. Use your tools to discover active markets on Polymarket
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,
        get_polymarket_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 Polymarket 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 Polymarket 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:
   - Token ID
   - Side (BUY/SELL)
   - Size (number of shares)
   - Limit price (between 0.01 and 0.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 Polymarket",
    system_prompt="""You are a Polymarket trade execution agent. Given approved trades from the risk manager:

1. Parse each approved trade's parameters
2. Use the place_polymarket_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_polymarket_order],
    output_type="str",
)

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

result = swarm.run(
    "Scan Polymarket 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(
    "polymarket-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,
}))