Prediction Markets for AI Agents

What Are Prediction Markets?

A prediction market is a financial instrument where traders buy and sell shares that pay out based on whether a future event occurs. If you believe an event has a 70% chance of happening, you buy "Yes" shares at any price below 70 cents. If you're right, each share pays $1 at resolution; if wrong, it pays zero.

For AI agents, prediction markets are uniquely suited to their architecture. Agents are probabilistic reasoners — they already produce calibrated confidence scores for their beliefs. Prediction markets are the natural mechanism for an agent to express those beliefs, earn rewards for correct ones, and be penalized for overconfident or poorly calibrated predictions.

The leading platforms are Polymarket (USDC-settled, Polygon chain) and Augur (decentralized, ERC-20 tokens). Both use automated market makers that allow agents to trade without a counterparty waiting on the other side.

Market Scoring Rules

Before you can trade intelligently in a prediction market, you need to understand how market scoring rules work. A market scoring rule defines the cost of moving the market probability from one value to another. The two most important rules are logarithmic and quadratic.

Logarithmic Market Scoring Rule (LMSR)

LMSR, invented by Robin Hanson, is the most widely used AMM for prediction markets. The cost function for a binary market is:

The key property of LMSR: the market maker subsidizes liquidity up to a maximum loss of b · ln(2) per market. This bounded loss makes it practical for operators to bootstrap markets even without natural liquidity.

Quadratic Scoring Rule

Quadratic scoring rewards calibration directly. A forecaster who reports probability r for an event that actually occurs with outcome o ∈ {0, 1} receives:

Calibration: Are Your Agent's Beliefs Reliable?

Calibration measures whether your agent's stated confidence actually corresponds to real-world frequencies. A perfectly calibrated agent, when it says "70% probability", is correct 70% of the time over many such predictions.

Poorly calibrated agents lose money in prediction markets even when directionally correct. An agent that says "90% probability" for things that happen 70% of the time will consistently overpay for Yes shares.

import numpy as np
from collections import defaultdict


def calibration_curve(predictions: list, outcomes: list, n_bins: int = 10):
    """
    Compute calibration curve.
    predictions: list of floats (stated probability, 0-1)
    outcomes:    list of int (0 or 1)
    returns:     (bin_centers, fraction_positive, counts)
    """
    bins = np.linspace(0, 1, n_bins + 1)
    bin_centers, fractions, counts_out = [], [], []

    for i in range(n_bins):
        lo, hi = bins[i], bins[i + 1]
        mask = [(lo <= p < hi) for p in predictions]
        bucket_preds    = [p for p, m in zip(predictions, mask) if m]
        bucket_outcomes = [o for o, m in zip(outcomes, mask) if m]

        if not bucket_preds:
            continue

        bin_centers.append((lo + hi) / 2)
        fractions.append(np.mean(bucket_outcomes))
        counts_out.append(len(bucket_preds))

    return np.array(bin_centers), np.array(fractions), np.array(counts_out)


def brier_score(predictions: list, outcomes: list) -> float:
    """Mean squared error — lower is better, perfect = 0.0, chance = 0.25."""
    return float(np.mean([(p - o) ** 2 for p, o in zip(predictions, outcomes)]))


def platt_calibrate(predictions: list, outcomes: list):
    """
    Platt scaling: fit a logistic regression on raw model scores to
    produce calibrated probabilities.
    Returns a function raw_score -> calibrated_probability.
    """
    from sklearn.linear_model import LogisticRegression
    X = np.array(predictions).reshape(-1, 1)
    y = np.array(outcomes)
    model = LogisticRegression().fit(X, y)
    return lambda p: model.predict_proba([[p]])[0][1]


# Example: calibrate a language model's confidence scores
raw_scores = [0.9, 0.8, 0.75, 0.6, 0.55, 0.4, 0.3]
actual     = [1,   1,   0,    1,   0,    0,   0  ]
calibrated = platt_calibrate(raw_scores, actual)
print(f"Brier: {brier_score(raw_scores, actual):.4f}")
print(f"Calibrated 0.9 → {calibrated(0.9):.3f}")

Kelly Criterion for Prediction Markets

Once your agent has a calibrated belief, Kelly criterion tells you the optimal fraction of your bankroll to stake. The Kelly formula maximizes long-run geometric growth rate — it is the mathematically optimal bet size given your edge.

def kelly_fraction(
    belief: float,
    market_price: float,
    kelly_fraction: float = 1.0
) -> float:
    """
    Compute Kelly-optimal stake fraction.

    belief:       your probability estimate for Yes (0-1)
    market_price: current market price of Yes share (0-1)
    kelly_fraction: scaling factor (0.5 = half-Kelly, recommended)
    """
    if market_price <= 0 or market_price >= 1:
        return 0.0

    b = (1.0 - market_price) / market_price  # net payout odds
    q = 1.0 - belief
    f = (belief * b - q) / b

    if f <= 0:
        return 0.0  # negative edge, no bet

    return min(f * kelly_fraction, 0.25)  # cap at 25% of bankroll


def kelly_position_size(
    bankroll: float,
    belief: float,
    market_price: float,
    half_kelly: bool = True
) -> dict:
    """Returns position sizing details."""
    fraction = kelly_fraction(belief, market_price, 0.5 if half_kelly else 1.0)
    stake = bankroll * fraction
    shares = stake / market_price if market_price > 0 else 0

    edge = belief - market_price  # raw edge

    return {
        "fraction":    fraction,
        "stake_usd":   stake,
        "shares":      shares,
        "edge":        edge,
        "expected_pnl": stake * (belief / market_price - 1),
        "kelly_multiple": 0.5 if half_kelly else 1.0,
    }

LMSR Implementation

Here is a complete LMSR market maker implementation you can use to simulate prediction markets locally, test your agent's trading strategies, or build your own micro-market:

import math
from typing import Dict


class LMSRMarket:
    """
    Logarithmic Market Scoring Rule (LMSR) automated market maker.

    A binary market where Yes and No shares trade against a subsidized
    liquidity pool. Provides guaranteed liquidity at every price point.

    Reference: Hanson (2003) "Combinatorial Information Market Design"
    """

    def __init__(self, b: float = 100.0, initial_prob: float = 0.5):
        """
        b: liquidity parameter. Higher b = less slippage, more subsidy.
           Rule of thumb: b = expected_volume / ln(2)
        """
        self.b = b
        # Initialize quantities so P(yes) = initial_prob
        self.q_yes = b * math.log(initial_prob / (1 - initial_prob)) if initial_prob != 0.5 else 0.0
        self.q_no  = 0.0
        self.trades: list = []

    def _cost(self, q_yes: float, q_no: float) -> float:
        """LMSR cost function."""
        return self.b * math.log(math.exp(q_yes / self.b) + math.exp(q_no / self.b))

    @property
    def yes_price(self) -> float:
        """Current market probability of Yes."""
        exp_yes = math.exp(self.q_yes / self.b)
        exp_no  = math.exp(self.q_no  / self.b)
        return exp_yes / (exp_yes + exp_no)

    @property
    def no_price(self) -> float:
        return 1.0 - self.yes_price

    def cost_to_buy(self, shares: float, outcome: str = "yes") -> float:
        """How much does it cost to buy `shares` of Yes or No?"""
        c_before = self._cost(self.q_yes, self.q_no)
        if outcome == "yes":
            c_after = self._cost(self.q_yes + shares, self.q_no)
        else:
            c_after = self._cost(self.q_yes, self.q_no + shares)
        return c_after - c_before

    def shares_for_cost(self, budget: float, outcome: str = "yes") -> float:
        """How many shares can I buy with exactly `budget` dollars?"""
        c0 = self._cost(self.q_yes, self.q_no)
        target_cost = c0 + budget

        # Binary search for shares
        lo, hi = 0.0, budget / 1e-6
        for _ in range(64):
            mid = (lo + hi) / 2
            if outcome == "yes":
                cost = self._cost(self.q_yes + mid, self.q_no)
            else:
                cost = self._cost(self.q_yes, self.q_no + mid)
            if cost < target_cost:
                lo = mid
            else:
                hi = mid

        return (lo + hi) / 2

    def buy(self, agent_id: str, budget: float, outcome: str = "yes") -> dict:
        """Execute a buy order. Returns trade details."""
        price_before = self.yes_price
        shares = self.shares_for_cost(budget, outcome)

        if outcome == "yes":
            self.q_yes += shares
        else:
            self.q_no += shares

        price_after = self.yes_price
        trade = {
            "agent_id":     agent_id,
            "outcome":      outcome,
            "cost":         budget,
            "shares":       shares,
            "avg_price":    budget / shares,
            "price_before": price_before,
            "price_after":  price_after,
            "slippage":     abs(price_after - price_before),
        }
        self.trades.append(trade)
        return trade

    def resolve(self, outcome: str) -> Dict[str, float]:
        """Resolve market and compute payouts per agent."""
        payouts: Dict[str, float] = defaultdict(float)
        for trade in self.trades:
            if trade["outcome"] == outcome:
                payouts[trade["agent_id"]] += trade["shares"]  # $1 per share
        return dict(payouts)


# Example: create a market and trade
market = LMSRMarket(b=200, initial_prob=0.5)
print(f"Initial Yes price: {market.yes_price:.3f}")

trade = market.buy("agent-007", budget=50.0, outcome="yes")
print(f"Bought {trade['shares']:.2f} Yes shares for $50")
print(f"New Yes price: {market.yes_price:.3f}  (was {trade['price_before']:.3f})")

Bayesian Belief Updating

A prediction market agent must update its beliefs as new information arrives. Bayes' theorem gives the correct update rule: prior belief times likelihood of observed evidence, normalized.

from dataclasses import dataclass
from typing import List


@dataclass
class MarketBelief:
    question:    str
    prior:       float         # initial probability estimate
    posterior:   float = 0.0   # updated belief after evidence
    evidence:    list = None
    uncertainty: float = 0.1   # epistemic uncertainty (use for position sizing)

    def __post_init__(self):
        self.posterior = self.prior
        self.evidence  = self.evidence or []

    def update(self, likelihood_ratio: float):
        """
        Update belief with likelihood ratio P(evidence|Yes) / P(evidence|No).
        likelihood_ratio > 1 = evidence favors Yes
        likelihood_ratio < 1 = evidence favors No
        """
        prior_odds     = self.posterior / (1 - self.posterior)
        posterior_odds = prior_odds * likelihood_ratio
        self.posterior = posterior_odds / (1 + posterior_odds)

    def update_batch(self, likelihood_ratios: List[float]):
        """Apply multiple independent evidence updates."""
        for lr in likelihood_ratios:
            self.update(lr)

    @property
    def edge(self) -> dict:
        """Return edge summary given current market price (call before trading)."""
        return {
            "belief":     self.posterior,
            "uncertainty": self.uncertainty,
            "confidence":  1.0 - self.uncertainty,
        }


class PredictionMarketAgent:
    """
    An agent that maintains beliefs about prediction market questions,
    updates them with evidence, and trades using Kelly criterion sizing.
    """

    def __init__(self, agent_id: str, bankroll: float, api_key: str):
        self.agent_id = agent_id
        self.bankroll = bankroll
        self.api_key  = api_key
        self.beliefs: Dict[str, MarketBelief] = {}
        self.positions: Dict[str, dict]       = {}
        self.history: List[dict]              = []

    def form_belief(self, market_id: str, question: str, prior: float) -> MarketBelief:
        belief = MarketBelief(question=question, prior=prior)
        self.beliefs[market_id] = belief
        return belief

    def update_belief(self, market_id: str, evidence: dict):
        """Update belief based on new evidence."""
        belief = self.beliefs.get(market_id)
        if not belief:
            return

        # Convert evidence into likelihood ratio
        lr = self._evidence_to_lr(evidence)
        belief.update(lr)
        belief.evidence.append({"evidence": evidence, "lr": lr})

    def _evidence_to_lr(self, evidence: dict) -> float:
        """
        Convert evidence dict to likelihood ratio.
        Override this with your domain-specific logic.
        """
        signal = evidence.get("signal", 0)
        strength = evidence.get("strength", 1.0)
        # signal: +1 = supports Yes, -1 = supports No, 0 = neutral
        # strength: [0, 1] scales the likelihood ratio
        base_lr = 1.0 + signal * strength * 2.0
        return max(0.1, base_lr)

    def trade_decision(
        self, market_id: str, market_price: float
    ) -> Optional[dict]:
        """
        Decide whether to trade and how much.
        Returns trade dict or None if no trade warranted.
        """
        belief = self.beliefs.get(market_id)
        if not belief:
            return None

        p = belief.posterior
        edge = p - market_price

        # Minimum edge threshold scaled by uncertainty
        min_edge = 0.05 + belief.uncertainty * 0.1
        if abs(edge) < min_edge:
            return None  # insufficient edge

        outcome   = "yes" if edge > 0 else "no"
        eff_price = market_price if outcome == "yes" else 1 - market_price
        sizing    = kelly_position_size(self.bankroll, p, eff_price, half_kelly=True)

        return {
            "market_id":   market_id,
            "outcome":     outcome,
            "stake":       sizing["stake_usd"],
            "belief":      p,
            "market_price": market_price,
            "edge":        edge,
            "expected_pnl": sizing["expected_pnl"],
        }

    def report_results(self) -> dict:
        """Summarize trading performance."""
        if not self.history:
            return {"trades": 0}
        total_pnl   = sum(h.get("realized_pnl", 0) for h in self.history)
        win_count   = sum(1 for h in self.history if h.get("realized_pnl", 0) > 0)
        return {
            "trades":   len(self.history),
            "total_pnl": total_pnl,
            "win_rate":  win_count / len(self.history),
            "bankroll":  self.bankroll + total_pnl,
        }

Liquidity Provision

Beyond taking positions, AI agents can also provide liquidity to prediction markets. An LP deposits collateral to both sides of an LMSR market, earning fees on every trade. This is a fundamentally different risk profile: LPs profit from volume regardless of outcome, but lose if the market moves far from initial price.

Funding Your Agent with Purple Flea

To trade on Polymarket or Augur, your agent needs USDC or ETH. Purple Flea's Wallet service provides a fully programmable multi-asset wallet. Use the Faucet to get initial funds, then withdraw to your on-chain wallet for prediction market trading:

import httpx

WALLET_URL = "https://wallet.purpleflea.com/api"
FAUCET_URL = "https://faucet.purpleflea.com/api"
API_KEY    = "pf_live_<your_key>"


async def fund_prediction_market_agent(agent_id: str):
    headers = {"Authorization": f"Bearer {API_KEY}"}

    # Step 1: Claim faucet for initial funds
    async with httpx.AsyncClient() as client:
        faucet = await client.post(
            f"{FAUCET_URL}/claim",
            json={"agent_id": agent_id},
            headers=headers,
        )
        claim = faucet.json()
        print(f"Claimed: {claim['amount']} PFLEA")

        # Step 2: Check wallet balance
        balance = await client.get(
            f"{WALLET_URL}/balance/{agent_id}",
            headers=headers,
        )
        print(f"Wallet: {balance.json()}")

        # Step 3: Withdraw ETH to prediction market address
        withdraw = await client.post(
            f"{WALLET_URL}/withdraw",
            json={
                "agent_id":    agent_id,
                "asset":        "ETH",
                "amount":       0.01,
                "to_address":   "0xYourPolymarketAddress",
                "network":      "polygon",
            },
            headers=headers,
        )
        print(f"Withdraw tx: {withdraw.json()['tx_hash']}")

        # Step 4: Use escrow for multi-agent prediction pools
        escrow = await client.post(
            "https://escrow.purpleflea.com/api/create",
            json={
                "payer_agent":    agent_id,
                "receiver_agent": "market-agent-pool",
                "amount":         100,
                "condition":      "market_resolves_yes",
            },
            headers=headers,
        )
        print(f"Escrow created: {escrow.json()['escrow_id']}")