How to Build a Prediction Market Agent from Scratch

pip install -r requirements.txt

# agent/config.py
"""AgentBets.ai - Prediction Market Agent Configuration"""

import os
from dotenv import load_dotenv
from dataclasses import dataclass

load_dotenv()

@dataclass
class AgentConfig:
    """Core configuration for the arbitrage agent."""

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

    # --- Arbitrage Parameters ---
    MIN_SPREAD_PCT: float = 0.03      # Minimum 3% spread to trigger (after fees)
    MAX_POSITION_USD: float = 50.0    # Max USD per side of any single arb trade
    MIN_LIQUIDITY_USD: float = 500.0  # Skip markets with < $500 orderbook depth

    # --- Risk Management ---
    SESSION_BUDGET_USD: float = 200.0  # Total max spend per agent session
    DAILY_LOSS_LIMIT_USD: float = 100.0
    MAX_OPEN_POSITIONS: int = 5

    # --- Polling ---
    POLL_INTERVAL_SEC: int = 30        # How often to scan for opportunities

    # --- API Credentials (loaded from .env) ---
    KALSHI_API_KEY: str = os.getenv("KALSHI_API_KEY", "")
    KALSHI_API_SECRET: str = os.getenv("KALSHI_API_SECRET", "")

    # --- Wallet (Coinbase Agentic Wallet) ---
    CDP_API_KEY: str = os.getenv("CDP_API_KEY", "")
    CDP_API_SECRET: str = os.getenv("CDP_API_SECRET", "")


config = AgentConfig()

# agent/data_fetcher.py
"""Market data fetchers for Polymarket and Kalshi."""

import aiohttp
import asyncio
from dataclasses import dataclass
from typing import Optional
from agent.config import config


@dataclass
class MarketSnapshot:
    """Normalized market data from any platform."""
    platform: str           # 'polymarket' or 'kalshi'
    event_slug: str         # Human-readable event identifier
    market_id: str          # Platform-specific market ID
    yes_price: float        # Current YES price (0.0 to 1.0)
    no_price: float         # Current NO price (0.0 to 1.0)
    yes_liquidity: float    # USD depth on YES side
    no_liquidity: float     # USD depth on NO side
    last_updated: str       # ISO timestamp


class PolymarketFetcher:
    """Fetch market data from Polymarket CLOB API."""

    def __init__(self):
        self.base_url = config.POLYMARKET_API

    async def get_markets(self, limit: int = 50) -> list[dict]:
        """Fetch active markets from Polymarket."""
        async with aiohttp.ClientSession() as session:
            url = f"{self.base_url}/markets"
            params = {"limit": limit, "active": "true"}
            async with session.get(url, params=params) as resp:
                if resp.status == 200:
                    return await resp.json()
                return []

    async def get_orderbook(self, token_id: str) -> dict:
        """Fetch orderbook for a specific token (YES or NO)."""
        async with aiohttp.ClientSession() as session:
            url = f"{self.base_url}/book"
            params = {"token_id": token_id}
            async with session.get(url, params=params) as resp:
                if resp.status == 200:
                    return await resp.json()
                return {}

    def parse_market(self, raw: dict) -> Optional[MarketSnapshot]:
        """Convert raw Polymarket data to normalized snapshot."""
        try:
            tokens = raw.get("tokens", [])
            if len(tokens) < 2:
                return None

            yes_price = float(tokens[0].get("price", 0))
            no_price = float(tokens[1].get("price", 0))

            return MarketSnapshot(
                platform="polymarket",
                event_slug=raw.get("question", "unknown"),
                market_id=raw.get("condition_id", ""),
                yes_price=yes_price,
                no_price=no_price,
                yes_liquidity=float(raw.get("volume", 0)),
                no_liquidity=float(raw.get("volume", 0)),
                last_updated=raw.get("end_date_iso", ""),
            )
        except (KeyError, ValueError):
            return None


class KalshiFetcher:
    """Fetch market data from Kalshi REST API."""

    def __init__(self):
        self.base_url = config.KALSHI_API
        self.api_key = config.KALSHI_API_KEY

    async def get_markets(
        self, status: str = "open", limit: int = 50
    ) -> list[dict]:
        """Fetch active markets from Kalshi."""
        headers = {"Authorization": f"Bearer {self.api_key}"}
        async with aiohttp.ClientSession() as session:
            url = f"{self.base_url}/markets"
            params = {"status": status, "limit": limit}
            async with session.get(
                url, headers=headers, params=params
            ) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return data.get("markets", [])
                return []

    def parse_market(self, raw: dict) -> Optional[MarketSnapshot]:
        """Convert raw Kalshi data to normalized snapshot."""
        try:
            yes_price = raw.get("yes_ask", 0) / 100.0
            no_price = raw.get("no_ask", 0) / 100.0

            return MarketSnapshot(
                platform="kalshi",
                event_slug=raw.get("title", "unknown"),
                market_id=raw.get("ticker", ""),
                yes_price=yes_price,
                no_price=no_price,
                yes_liquidity=float(raw.get("volume", 0)),
                no_liquidity=float(raw.get("volume", 0)),
                last_updated=raw.get("close_time", ""),
            )
        except (KeyError, ValueError, ZeroDivisionError):
            return None

# agent/arbitrage.py
"""Arbitrage detection engine for cross-platform prediction markets."""

from dataclasses import dataclass
from typing import Optional
from agent.data_fetcher import MarketSnapshot
from agent.config import config


EVENT_MAP: dict[str, str] = {
    # "polymarket_condition_id": "kalshi_ticker"
    # Example: "0x1234...abcd": "PRES-2028-DEM"
}


@dataclass
class ArbitrageOpportunity:
    """A detected cross-platform arbitrage opportunity."""
    event_name: str
    platform_a: str
    platform_b: str
    buy_yes_on: str
    buy_no_on: str
    yes_price: float
    no_price: float
    total_cost: float
    gross_spread: float
    net_spread: float
    estimated_roi: float


def estimate_fees(cost: float) -> float:
    """
    Estimate total round-trip fees for both platforms.
    Polymarket: ~2% effective fee (maker/taker on CLOB)
    Kalshi: ~1-7 cents per contract depending on price
    This is a rough approximation. Always check current fee schedules.
    """
    return cost * 0.025  # 2.5% combined estimate


def detect_arbitrage(
    market_a: MarketSnapshot,
    market_b: MarketSnapshot,
) -> Optional[ArbitrageOpportunity]:
    """
    Check if two matched markets have an arbitrage spread.
    Returns an ArbitrageOpportunity if the net spread exceeds
    the configured MIN_SPREAD_PCT threshold, otherwise None.
    """

    # Strategy 1: Buy YES on A, buy NO on B
    cost_1 = market_a.yes_price + market_b.no_price
    gross_1 = 1.0 - cost_1
    fees_1 = estimate_fees(cost_1)
    net_1 = gross_1 - fees_1

    # Strategy 2: Buy YES on B, buy NO on A
    cost_2 = market_b.yes_price + market_a.no_price
    gross_2 = 1.0 - cost_2
    fees_2 = estimate_fees(cost_2)
    net_2 = gross_2 - fees_2

    # Pick the better strategy
    if net_1 >= net_2 and net_1 > 0:
        best_cost, best_net = cost_1, net_1
        buy_yes = market_a.platform
        buy_no = market_b.platform
        yes_px = market_a.yes_price
        no_px = market_b.no_price
    elif net_2 > 0:
        best_cost, best_net = cost_2, net_2
        buy_yes = market_b.platform
        buy_no = market_a.platform
        yes_px = market_b.yes_price
        no_px = market_a.no_price
    else:
        return None

    roi = best_net / best_cost if best_cost > 0 else 0

    if roi < config.MIN_SPREAD_PCT:
        return None

    if (market_a.yes_liquidity < config.MIN_LIQUIDITY_USD or
            market_b.no_liquidity < config.MIN_LIQUIDITY_USD):
        return None

    return ArbitrageOpportunity(
        event_name=market_a.event_slug,
        platform_a=market_a.platform,
        platform_b=market_b.platform,
        buy_yes_on=buy_yes,
        buy_no_on=buy_no,
        yes_price=yes_px,
        no_price=no_px,
        total_cost=best_cost,
        gross_spread=1.0 - best_cost,
        net_spread=best_net,
        estimated_roi=roi,
    )

# agent/risk_manager.py
"""Risk management and spending controls for the arbitrage agent."""

import time
import logging
from dataclasses import dataclass, field
from agent.config import config
from agent.arbitrage import ArbitrageOpportunity

logger = logging.getLogger(__name__)


@dataclass
class RiskState:
    """Tracks session-level risk metrics."""
    session_spent: float = 0.0
    daily_loss: float = 0.0
    open_positions: int = 0
    trades_executed: int = 0
    session_start: float = field(default_factory=time.time)
    is_killed: bool = False


class RiskManager:
    """Enforces spending limits and risk guardrails."""

    def __init__(self):
        self.state = RiskState()

    def kill_switch(self, reason: str) -> None:
        """Emergency shutdown. Prevents all further trading."""
        self.state.is_killed = True
        logger.critical(
            f'KILL SWITCH ACTIVATED: {reason}. '
            f'Session spent: ${self.state.session_spent:.2f}. '
            f'Trades: {self.state.trades_executed}'
        )

    def can_trade(self, opp: ArbitrageOpportunity) -> tuple[bool, str]:
        """
        Check if a proposed trade passes all risk checks.
        Returns (allowed: bool, reason: str).
        """
        if self.state.is_killed:
            return False, "Kill switch is active"

        trade_cost = opp.total_cost * config.MAX_POSITION_USD

        if self.state.session_spent + trade_cost > config.SESSION_BUDGET_USD:
            return False, (
                f'Session budget exceeded: '
                f'${self.state.session_spent:.2f} + '
                f'${trade_cost:.2f} > '
                f'${config.SESSION_BUDGET_USD:.2f}'
            )

        if self.state.daily_loss >= config.DAILY_LOSS_LIMIT_USD:
            self.kill_switch("Daily loss limit reached")
            return False, "Daily loss limit reached"

        if self.state.open_positions >= config.MAX_OPEN_POSITIONS:
            return False, (
                f'Max open positions: {self.state.open_positions}'
            )

        if trade_cost > config.MAX_POSITION_USD:
            return False, (
                f'Trade exceeds per-position limit: '
                f'${trade_cost:.2f} > ${config.MAX_POSITION_USD:.2f}'
            )

        if opp.estimated_roi < config.MIN_SPREAD_PCT:
            return False, "ROI below threshold after re-check"

        return True, "All checks passed"

    def record_trade(self, cost: float) -> None:
        """Record a completed trade."""
        self.state.session_spent += cost
        self.state.trades_executed += 1
        self.state.open_positions += 1
        logger.info(
            f'Trade recorded. Session total: '
            f'${self.state.session_spent:.2f} / '
            f'${config.SESSION_BUDGET_USD:.2f}'
        )

    def record_loss(self, amount: float) -> None:
        """Record a realized loss."""
        self.state.daily_loss += amount
        if self.state.daily_loss >= config.DAILY_LOSS_LIMIT_USD:
            self.kill_switch("Daily loss limit hit")

# agent/wallet.py
"""Wallet abstraction for the arbitrage agent."""

import logging
from dataclasses import dataclass
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class TradeOrder:
    """A trade to execute on a specific platform."""
    platform: str         # 'polymarket' or 'kalshi'
    market_id: str
    side: str             # 'YES' or 'NO'
    amount_usd: float
    max_price: float      # Limit price (slippage protection)


@dataclass
class TradeResult:
    """Result of an attempted trade execution."""
    success: bool
    order_id: Optional[str] = None
    filled_price: Optional[float] = None
    filled_amount: Optional[float] = None
    error: Optional[str] = None


class WalletManager:
    """
    Abstraction over wallet operations.

    In production, this connects to Coinbase Agentic Wallet SDK.
    For this educational build, it logs intended trades without
    executing them (paper trading).
    """

    def __init__(self, paper_mode: bool = True):
        self.paper_mode = paper_mode
        self.balance = 0.0

    async def get_balance(self) -> float:
        """Get current USDC balance."""
        if self.paper_mode:
            return 1000.0  # Simulated balance
        raise NotImplementedError("Connect to wallet SDK")

    async def execute_trade(self, order: TradeOrder) -> TradeResult:
        """
        Execute a trade on the specified platform.

        In paper mode, this simulates execution.
        In live mode, this would:
        1. Sign the transaction with the enclave key
        2. Submit to the platform's CLOB/API
        3. Monitor for fill confirmation
        4. Return the result
        """
        if self.paper_mode:
            logger.info(
                f'[PAPER] {order.side} on {order.platform}: '
                f'${order.amount_usd:.2f} at max {order.max_price:.4f} '
                f'for market {order.market_id}'
            )
            return TradeResult(
                success=True,
                order_id="paper-" + order.market_id[:8],
                filled_price=order.max_price,
                filled_amount=order.amount_usd,
            )
        raise NotImplementedError(
            "Live trading requires Coinbase Agentic Wallet SDK. "
            "See: /guides/coinbase-agentic-wallets-guide/"
        )

    async def execute_arb_pair(
        self, yes_order: TradeOrder, no_order: TradeOrder
    ) -> tuple[TradeResult, TradeResult]:
        """
        Execute both legs of an arbitrage trade.

        CRITICAL: In production, implement atomic execution.
        If the first leg fills but the second fails, you are
        exposed to directional risk.
        """
        result_a = await self.execute_trade(yes_order)

        if not result_a.success:
            return result_a, TradeResult(
                success=False, error="Skipped: first leg failed"
            )

        result_b = await self.execute_trade(no_order)

        if not result_b.success:
            logger.warning(
                "PARTIAL FILL: First leg succeeded, second failed. "
                "Manual intervention may be required."
            )

        return result_a, result_b

# agent/main.py
"""Main orchestrator for the prediction market arbitrage agent."""

import asyncio
import logging
from agent.config import config
from agent.data_fetcher import PolymarketFetcher, KalshiFetcher
from agent.arbitrage import detect_arbitrage, EVENT_MAP
from agent.risk_manager import RiskManager
from agent.wallet import WalletManager, TradeOrder

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger(__name__)


async def run_scan_cycle(
    poly: PolymarketFetcher,
    kalshi: KalshiFetcher,
    risk: RiskManager,
    wallet: WalletManager,
) -> None:
    """Run one scan cycle across all mapped events."""

    logger.info("--- Starting scan cycle ---")

    poly_markets_raw = await poly.get_markets(limit=100)
    kalshi_markets_raw = await kalshi.get_markets(limit=100)

    poly_markets = {}
    for raw in poly_markets_raw:
        snapshot = poly.parse_market(raw)
        if snapshot:
            poly_markets[snapshot.market_id] = snapshot

    kalshi_markets = {}
    for raw in kalshi_markets_raw:
        snapshot = kalshi.parse_market(raw)
        if snapshot:
            kalshi_markets[snapshot.market_id] = snapshot

    logger.info(
        f'Fetched {len(poly_markets)} Polymarket, '
        f'{len(kalshi_markets)} Kalshi markets'
    )

    opportunities_found = 0
    for poly_id, kalshi_ticker in EVENT_MAP.items():
        poly_snap = poly_markets.get(poly_id)
        kalshi_snap = kalshi_markets.get(kalshi_ticker)

        if not poly_snap or not kalshi_snap:
            continue

        opp = detect_arbitrage(poly_snap, kalshi_snap)
        if opp is None:
            continue

        opportunities_found += 1
        logger.info(
            f'ARB DETECTED: {opp.event_name} | '
            f'Buy YES@{opp.buy_yes_on} ${opp.yes_price:.4f} + '
            f'Buy NO@{opp.buy_no_on} ${opp.no_price:.4f} | '
            f'Cost: ${opp.total_cost:.4f} | '
            f'Net spread: ${opp.net_spread:.4f} | '
            f'ROI: {opp.estimated_roi:.2%}'
        )

        allowed, reason = risk.can_trade(opp)
        if not allowed:
            logger.info(f'Trade blocked by risk manager: {reason}')
            continue

        position_size = min(
            config.MAX_POSITION_USD,
            config.SESSION_BUDGET_USD - risk.state.session_spent,
        )

        yes_order = TradeOrder(
            platform=opp.buy_yes_on,
            market_id=(
                poly_id if opp.buy_yes_on == "polymarket"
                else kalshi_ticker
            ),
            side="YES",
            amount_usd=position_size * opp.yes_price,
            max_price=opp.yes_price * 1.005,
        )
        no_order = TradeOrder(
            platform=opp.buy_no_on,
            market_id=(
                kalshi_ticker if opp.buy_no_on == "kalshi"
                else poly_id
            ),
            side="NO",
            amount_usd=position_size * opp.no_price,
            max_price=opp.no_price * 1.005,
        )

        res_a, res_b = await wallet.execute_arb_pair(
            yes_order, no_order
        )
        if res_a.success and res_b.success:
            trade_cost = (
                (res_a.filled_amount or 0) +
                (res_b.filled_amount or 0)
            )
            risk.record_trade(trade_cost)
            logger.info(f'Trade executed. Cost: ${trade_cost:.2f}')

    logger.info(
        f'Scan complete. Found {opportunities_found} opportunities. '
        f'Session spent: ${risk.state.session_spent:.2f}'
    )


async def main():
    """Main agent loop."""
    logger.info("=" * 60)
    logger.info("AgentBets.ai Arbitrage Agent Starting")
    logger.info("Mode: PAPER TRADING")
    logger.info("=" * 60)

    poly = PolymarketFetcher()
    kalshi = KalshiFetcher()
    risk = RiskManager()
    wallet = WalletManager(paper_mode=True)

    while not risk.state.is_killed:
        try:
            await run_scan_cycle(poly, kalshi, risk, wallet)
        except Exception as e:
            logger.error(f'Scan cycle error: {e}')

        logger.info(
            f'Sleeping {config.POLL_INTERVAL_SEC}s until next scan...'
        )
        await asyncio.sleep(config.POLL_INTERVAL_SEC)

    logger.info("Agent shut down.")


if __name__ == "__main__":
    asyncio.run(main())

# tests/test_arbitrage.py
"""Unit tests for the arbitrage detection engine."""

import pytest
from agent.data_fetcher import MarketSnapshot
from agent.arbitrage import detect_arbitrage


def make_snapshot(platform, yes, no, liq=1000.0):
    return MarketSnapshot(
        platform=platform,
        event_slug="Test Event",
        market_id="test-123",
        yes_price=yes,
        no_price=no,
        yes_liquidity=liq,
        no_liquidity=liq,
        last_updated="2026-03-05T00:00:00Z",
    )


def test_detects_clear_arb():
    """When combined cost < 1.0 minus fees, arb should be detected."""
    a = make_snapshot("polymarket", yes=0.40, no=0.62)
    b = make_snapshot("kalshi", yes=0.55, no=0.48)
    opp = detect_arbitrage(a, b)
    assert opp is not None
    assert opp.estimated_roi > 0.03


def test_no_arb_when_efficient():
    """When prices are efficient, no arb should be detected."""
    a = make_snapshot("polymarket", yes=0.52, no=0.49)
    b = make_snapshot("kalshi", yes=0.51, no=0.50)
    opp = detect_arbitrage(a, b)
    assert opp is None


def test_no_arb_low_liquidity():
    """Arb should be rejected if liquidity is below threshold."""
    a = make_snapshot("polymarket", yes=0.40, no=0.62, liq=100)
    b = make_snapshot("kalshi", yes=0.55, no=0.48, liq=100)
    opp = detect_arbitrage(a, b)
    assert opp is None


def test_chooses_best_direction():
    """Agent should pick the more profitable direction."""
    a = make_snapshot("polymarket", yes=0.35, no=0.68)
    b = make_snapshot("kalshi", yes=0.60, no=0.42)
    opp = detect_arbitrage(a, b)
    assert opp is not None
    assert opp.buy_yes_on == "polymarket"
    assert opp.buy_no_on == "kalshi"