套利警報：使用 Python 建立跨交易所預測市場監察工具

Introduction

Prediction markets like Polymarket and Kalshi provide real-time probabilities for future events. On October 18, 2024, they priced the Fed's October rate decision four percentage points apart: Polymarket implied a 92% chance of a 25bp cut, while Kalshi showed 88%. Such discrepancies create arbitrage opportunities, but manually monitoring multiple platforms is impractical. In this post, we'll build a Python agent that polls both exchanges, normalizes contract prices, and computes the spread. The agent runs in a loop with explicit termination conditions and uses only public APIs. We'll also discuss security and precision issues.

Problem

Prediction markets often disagree due to differing liquidity, fees, or user bases. For developers building AI agents, detecting these spreads programmatically is a first step toward automated trading or alerting. Challenges include:

• API heterogeneity: Polymarket uses a GraphQL endpoint; Kalshi uses REST.
• Contract specifications: Outcomes may be phrased differently (e.g., "Fed cuts rates by 25bp" vs. "FOMC decision: cut 25bp").
• Real-time updates: Prices change frequently; polling must be efficient.
• Error handling: Network failures, rate limits, and malformed responses.

We'll focus on a minimal viable monitor that fetches current prices for a specific market and computes the spread. We'll use Python 3.9+ with requests and asyncio.

Solution

Our agent will:

1. Fetch the current price for a given market on Polymarket and Kalshi.
2. Normalize prices to probabilities (0-1).
3. Compute the absolute difference.
4. Log the spread if it exceeds a threshold.
5. Terminate after a specified number of iterations or when a stop condition is met.

We'll define an agent loop with explicit termination: it stops after max_iterations or if a KeyboardInterrupt occurs. We'll use asyncio for concurrent polling.

Security note: This code only reads public data. Do not use eval or exec on API responses. Always validate JSON schemas.

Implementation

Prerequisites

pip install requests aiohttp

Code

import asyncio
import aiohttp
import json
from typing import Optional, Dict, Any

# Configuration
POLYMARKET_API = "https://clob.polymarket.com"
KALSHI_API = "https://api.elections.kalshi.com/trade-api/v2"

# Example market IDs (replace with actual IDs from the platforms)
POLYMARKET_MARKET_ID = "0x123..."  # Polymarket condition ID
KALSHI_TICKER = "FED-24OCT-T25"    # Kalshi market ticker

async def fetch_polymarket_price(session: aiohttp.ClientSession, market_id: str) -> Optional[float]:
    """Fetch the yes price (probability) from Polymarket."""
    # Polymarket's CLOB API requires a token ID; this is a simplified example.
    # In practice, you'd query the market's token IDs first.
    url = f"{POLYMARKET_API}/price?token_id={market_id}&side=BUY"
    try:
        async with session.get(url, timeout=10) as resp:
            if resp.status != 200:
                return None
            data = await resp.json()
            # Response format: {"price": "0.92"}
            return float(data.get("price", 0))
    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError):
        return None

async def fetch_kalshi_price(session: aiohttp.ClientSession, ticker: str) -> Optional[float]:
    """Fetch the yes price from Kalshi."""
    url = f"{KALSHI_API}/markets/{ticker}"
    try:
        async with session.get(url, timeout=10) as resp:
            if resp.status != 200:
                return None
            data = await resp.json()
            # Kalshi returns prices in cents (0-100)
            yes_price = data.get("market", {}).get("yes_price")
            if yes_price is not None:
                return yes_price / 100.0
            return None
    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError):
        return None

async def monitor_spread(max_iterations: int = 10, interval: int = 5, threshold: float = 0.02):
    """Poll both exchanges and log spreads above threshold."""
    async with aiohttp.ClientSession() as session:
        for i in range(max_iterations):
            poly_price, kalshi_price = await asyncio.gather(
                fetch_polymarket_price(session, POLYMARKET_MARKET_ID),
                fetch_kalshi_price(session, KALSHI_TICKER)
            )
            if poly_price is not None and kalshi_price is not None:
                spread = abs(poly_price - kalshi_price)
                print(f"Iteration {i+1}: Polymarket={poly_price:.2%}, Kalshi={kalshi_price:.2%}, Spread={spread:.2%}")
                if spread >= threshold:
                    print(f"  ALERT: Spread exceeds {threshold:.0%}")
            else:
                print(f"Iteration {i+1}: Failed to fetch prices")
            await asyncio.sleep(interval)

if __name__ == "__main__":
    asyncio.run(monitor_spread(max_iterations=5, interval=10, threshold=0.03))

Explanation

• Polymarket: The CLOB API endpoint /price requires a token ID and side. In a real implementation, you'd first query /markets to get token IDs. We use a placeholder POLYMARKET_MARKET_ID.
• Kalshi: The /markets/{ticker} endpoint returns a yes_price in cents. We convert to probability by dividing by 100.
• Agent loop: The monitor_spread function runs for max_iterations (explicit termination). It uses asyncio.gather for concurrent fetching. Errors are caught and logged; the loop continues.
• Threshold: Alerts when spread >= 3% (default). Adjust based on fees and slippage.

Note: This is a simplified example. Actual market IDs and endpoints may change. Always refer to official API docs.

Key Takeaways

• Cross-exchange monitoring is feasible with public APIs, but requires careful normalization of contract specifications and price formats.
• Agent loops need termination conditions: we used max_iterations and asyncio.sleep to avoid infinite loops.
• Error handling is crucial: network issues and API changes can break your agent. Always validate responses and use timeouts.
• Security: Never use eval or exec on API data. Use json.loads and type checks.
• Precision: Prices may be strings; convert carefully. Use float only after validation.
• Arbitrage opportunities may be fleeting; consider building a more robust system with WebSockets for real-time updates.

This agent is a starting point. Extend it with alerting (e.g., email, Slack) and persistent storage for historical spreads.