Prediction market trading applies trading strategies, arbitrage detection, market making, and risk management techniques specifically to binary and categorical outcome markets. Unlike traditional financial markets, prediction markets have bounded payoffs (typically $0 or $1), guaranteed resolution dates, and outcomes tied to real-world events. These structural differences create unique opportunities for systematic trading, AI agent deployment (like the Olas/Polystrat approach), and portfolio construction across binary outcomes.

## Key Points

1. Base rates for similar events
2. Current evidence and trends
3. Key factors that could change the outcome
4. Potential biases in the current market price
1. Expected value in prediction markets is simply your_probability minus market_price; trade only when this edge exceeds your transaction costs and minimum threshold
2. Kelly Criterion gives optimal sizing but is very aggressive; use half-Kelly or quarter-Kelly in practice to survive variance
3. Cross-platform arbitrage requires prices to sum to less than 1 after fees; these opportunities are rare but risk-free when found
4. Market making in prediction markets earns the spread while managing inventory; skew quotes to flatten position when inventory grows
5. AI trading agents combine LLM research with systematic position sizing; the Olas/Polystrat approach automates the full pipeline from research to execution
6. Prediction market portfolios need Monte Carlo simulation because all positions are binary (no partial outcomes)
7. Correlation management is critical: political markets are often highly correlated, and a single wrong assessment can wipe out a concentrated portfolio
8. Pre-trade risk checks (position limits, drawdown limits, concentration limits, minimum edge) prevent the most common causes of ruin

Prediction Market Trading

Overview

Prediction market trading applies trading strategies, arbitrage detection, market making, and risk management techniques specifically to binary and categorical outcome markets. Unlike traditional financial markets, prediction markets have bounded payoffs (typically $0 or $1), guaranteed resolution dates, and outcomes tied to real-world events. These structural differences create unique opportunities for systematic trading, AI agent deployment (like the Olas/Polystrat approach), and portfolio construction across binary outcomes.

Trading Strategy Fundamentals

The Edge: When Markets Are Wrong

A prediction market trade is profitable when your estimated probability diverges from the market's implied probability:

import numpy as np

class PredictionMarketTrader:
    """Core trading logic for prediction markets."""

    def __init__(self, bankroll: float = 1000.0):
        self.bankroll = bankroll
        self.positions = {}  # contract_id -> {shares, avg_price, direction}
        self.trade_history = []

    def expected_value(self, market_price: float, your_probability: float,
                       direction: str = 'yes') -> dict:
        """
        Calculate expected value of a trade.

        Buy YES at price p when you think true probability is q:
          EV = q * (1 - p) - (1 - q) * p
             = q - p

        Edge = your_probability - market_price (for YES)
        """
        if direction == 'yes':
            cost = market_price
            profit_if_yes = 1 - market_price
            loss_if_no = market_price
            ev = your_probability * profit_if_yes - (1 - your_probability) * loss_if_no
            edge = your_probability - market_price
        else:  # Buying NO
            cost = 1 - market_price
            profit_if_no = market_price
            loss_if_yes = 1 - market_price
            ev = (1 - your_probability) * profit_if_no - your_probability * loss_if_yes
            edge = market_price - your_probability

        return {
            'expected_value_per_share': ev,
            'edge': edge,
            'cost_per_share': cost,
            'return_if_right': profit_if_yes if direction == 'yes' else profit_if_no,
            'loss_if_wrong': loss_if_no if direction == 'yes' else loss_if_yes,
            'should_trade': edge > 0.05,  # Minimum edge threshold
            'direction': direction
        }

    def kelly_criterion(self, probability: float, odds: float) -> float:
        """
        Kelly Criterion: optimal bet size to maximize long-term growth.

        For prediction markets with binary payoff:
        f* = (p * b - q) / b
        where p = your probability, q = 1-p, b = odds (payout ratio)

        Kelly is aggressive; most practitioners use fractional Kelly (25-50%).
        """
        q = 1 - probability
        kelly_fraction = (probability * odds - q) / odds

        # Fractional Kelly for safety
        half_kelly = kelly_fraction * 0.5

        return {
            'full_kelly': max(0, kelly_fraction),
            'half_kelly': max(0, half_kelly),
            'quarter_kelly': max(0, kelly_fraction * 0.25),
            'recommended_fraction': max(0, half_kelly),
            'bet_size': max(0, half_kelly * self.bankroll),
            'warning': 'Never risk more than 5% of bankroll on a single market' if kelly_fraction > 0.05 else ''
        }

    def size_position(self, market_price: float, your_probability: float,
                      direction: str = 'yes', max_pct: float = 0.05) -> dict:
        """
        Determine position size using Kelly + risk constraints.
        """
        ev = self.expected_value(market_price, your_probability, direction)

        if ev['edge'] <= 0:
            return {'shares': 0, 'reason': 'No edge'}

        cost_per_share = ev['cost_per_share']
        if direction == 'yes':
            odds = (1 - market_price) / market_price
        else:
            odds = market_price / (1 - market_price)

        kelly = self.kelly_criterion(your_probability if direction == 'yes' else 1 - your_probability, odds)

        # Apply maximum position constraint
        max_bet = self.bankroll * max_pct
        bet_size = min(kelly['bet_size'], max_bet)

        shares = int(bet_size / cost_per_share)

        return {
            'shares': shares,
            'total_cost': shares * cost_per_share,
            'pct_of_bankroll': (shares * cost_per_share) / self.bankroll,
            'expected_profit': shares * ev['expected_value_per_share'],
            'edge': ev['edge'],
            'kelly_fraction': kelly['half_kelly']
        }

    def execute_trade(self, contract_id: str, shares: int, price: float,
                      direction: str):
        """Record a trade execution."""
        cost = shares * (price if direction == 'yes' else 1 - price)

        if contract_id not in self.positions:
            self.positions[contract_id] = {
                'yes_shares': 0, 'no_shares': 0,
                'avg_yes_price': 0, 'avg_no_price': 0,
                'total_cost': 0
            }

        pos = self.positions[contract_id]
        if direction == 'yes':
            old_total = pos['yes_shares'] * pos['avg_yes_price']
            pos['yes_shares'] += shares
            pos['avg_yes_price'] = (old_total + cost) / pos['yes_shares'] if pos['yes_shares'] > 0 else 0
        else:
            old_total = pos['no_shares'] * pos['avg_no_price']
            pos['no_shares'] += shares
            pos['avg_no_price'] = (old_total + cost) / pos['no_shares'] if pos['no_shares'] > 0 else 0

        pos['total_cost'] += cost
        self.bankroll -= cost

        self.trade_history.append({
            'contract': contract_id,
            'direction': direction,
            'shares': shares,
            'price': price,
            'cost': cost,
            'bankroll_after': self.bankroll
        })

Arbitrage Detection

Cross-Market Arbitrage

class ArbitrageDetector:
    """Find arbitrage opportunities across prediction markets."""

    def __init__(self):
        self.markets = {}

    def add_market(self, platform: str, question: str, yes_price: float,
                   no_price: float, fees: float = 0.0):
        key = question.lower().strip()
        if key not in self.markets:
            self.markets[key] = []
        self.markets[key].append({
            'platform': platform,
            'yes_price': yes_price,
            'no_price': no_price,
            'fees': fees,
            'implied_prob': yes_price
        })

    def find_cross_platform_arb(self) -> list:
        """
        Find cases where the same question has different prices
        on different platforms.

        Example: Platform A has YES at 0.60, Platform B has NO at 0.50
        Buy YES on A for 0.60, buy NO on B for 0.50
        Total cost: 1.10 for guaranteed payout of 1.00... NO, that's a loss.

        Arb exists when: YES_A + NO_B < 1 (or YES_B + NO_A < 1)
        after fees.
        """
        opportunities = []

        for question, listings in self.markets.items():
            if len(listings) < 2:
                continue

            for i in range(len(listings)):
                for j in range(i + 1, len(listings)):
                    a = listings[i]
                    b = listings[j]

                    # Buy YES on A, buy NO on B
                    cost1 = a['yes_price'] + (1 - b['yes_price'])
                    cost1_after_fees = cost1 * (1 + max(a['fees'], b['fees']))

                    # Buy YES on B, buy NO on A
                    cost2 = b['yes_price'] + (1 - a['yes_price'])
                    cost2_after_fees = cost2 * (1 + max(a['fees'], b['fees']))

                    if cost1_after_fees < 1.0:
                        opportunities.append({
                            'question': question,
                            'strategy': f"Buy YES on {a['platform']}, NO on {b['platform']}",
                            'cost': cost1_after_fees,
                            'guaranteed_profit': 1.0 - cost1_after_fees,
                            'return_pct': (1.0 / cost1_after_fees - 1) * 100
                        })

                    if cost2_after_fees < 1.0:
                        opportunities.append({
                            'question': question,
                            'strategy': f"Buy YES on {b['platform']}, NO on {a['platform']}",
                            'cost': cost2_after_fees,
                            'guaranteed_profit': 1.0 - cost2_after_fees,
                            'return_pct': (1.0 / cost2_after_fees - 1) * 100
                        })

        return sorted(opportunities, key=lambda x: -x['return_pct'])

    def find_categorical_arb(self, outcomes: list, prices: list) -> dict:
        """
        For categorical markets (multiple mutually exclusive outcomes),
        prices should sum to ~1. If sum < 1, buy all. If sum > 1, short all.
        """
        total = sum(prices)
        fees_estimate = 0.02

        if total + fees_estimate < 1.0:
            return {
                'arb_exists': True,
                'type': 'buy_all',
                'total_cost': total,
                'guaranteed_profit': 1.0 - total - fees_estimate,
                'strategy': 'Buy one share of every outcome'
            }
        elif total - fees_estimate > 1.0:
            return {
                'arb_exists': True,
                'type': 'sell_all',
                'total_revenue': total,
                'guaranteed_profit': total - 1.0 - fees_estimate,
                'strategy': 'Sell one share of every outcome (if short selling available)'
            }

        return {'arb_exists': False, 'total': total}

Market Making

class PredictionMarketMaker:
    """
    Provide liquidity by continuously quoting bid and ask prices.
    Profit from the spread while managing inventory risk.
    """

    def __init__(self, initial_capital: float = 10000,
                 base_spread: float = 0.04):
        self.capital = initial_capital
        self.base_spread = base_spread
        self.inventory = 0  # Positive = net YES, negative = net NO
        self.max_inventory = 1000

    def quote(self, fair_value: float) -> dict:
        """
        Generate bid and ask prices.
        Adjust spread based on inventory to manage risk.
        """
        # Inventory-based skew: if we hold too many YES shares,
        # lower the bid (discourage more buying) and lower the ask
        inventory_skew = self.inventory / self.max_inventory * 0.02

        # Wider spread when uncertain or heavily positioned
        volatility_spread = self.base_spread
        inventory_spread = abs(self.inventory / self.max_inventory) * 0.02

        total_half_spread = (volatility_spread + inventory_spread) / 2

        bid = fair_value - total_half_spread - inventory_skew
        ask = fair_value + total_half_spread - inventory_skew

        # Bound to valid range
        bid = max(0.01, min(0.98, bid))
        ask = max(0.02, min(0.99, ask))

        return {
            'bid': round(bid, 3),
            'ask': round(ask, 3),
            'spread': round(ask - bid, 3),
            'fair_value': fair_value,
            'inventory': self.inventory,
            'inventory_skew': inventory_skew
        }

    def fill_order(self, side: str, shares: int, price: float):
        """Record a fill against our quotes."""
        if side == 'buy':  # Someone bought from us (we sold)
            self.inventory -= shares
            self.capital += shares * price
        else:  # Someone sold to us (we bought)
            self.inventory += shares
            self.capital -= shares * price

    def pnl_at_resolution(self, outcome: bool) -> float:
        """Calculate P&L when the market resolves."""
        if outcome:
            return self.capital + self.inventory * 1.0  # YES pays $1
        else:
            return self.capital  # YES pays $0

    def risk_metrics(self, fair_value: float) -> dict:
        """Current risk metrics."""
        mark_to_market = self.capital + self.inventory * fair_value
        max_loss_yes = self.capital + max(0, -self.inventory)  # If YES resolves
        max_loss_no = self.capital + max(0, self.inventory)    # If NO resolves

        return {
            'mark_to_market': mark_to_market,
            'worst_case': min(
                self.pnl_at_resolution(True),
                self.pnl_at_resolution(False)
            ),
            'inventory_value': abs(self.inventory) * fair_value,
            'inventory_exposure_pct': abs(self.inventory) * fair_value / mark_to_market * 100 if mark_to_market > 0 else 0
        }

AI Agent Trading (Olas/Polystrat Approach)

Autonomous Trading Agents

class AITradingAgent:
    """
    Autonomous AI agent for prediction market trading.
    Inspired by Olas (Autonolas) agents that trade on Polymarket.
    """

    def __init__(self, strategy: str, risk_budget: float = 1000):
        self.strategy = strategy
        self.trader = PredictionMarketTrader(bankroll=risk_budget)
        self.research_cache = {}
        self.active_markets = {}

    async def research_market(self, question: str, market_price: float) -> dict:
        """
        Use LLM to research a prediction market question.
        Returns an estimated probability with confidence.
        """
        research_prompt = f"""Analyze this prediction market question and estimate
the probability of YES resolution.

Question: {question}
Current market price (implied probability): {market_price:.1%}

Consider:
1. Base rates for similar events
2. Current evidence and trends
3. Key factors that could change the outcome
4. Potential biases in the current market price

Respond as JSON:
{{
    "probability": 0.XX,
    "confidence": "high/medium/low",
    "reasoning": "...",
    "key_factors": ["factor1", "factor2"],
    "information_sources": ["source1", "source2"]
}}"""

        # In production, call LLM here
        return {
            'probability': 0.5,
            'confidence': 'medium',
            'reasoning': 'placeholder',
            'key_factors': [],
            'information_sources': []
        }

    async def evaluate_opportunity(self, question: str, market_price: float,
                                    volume: float, time_to_resolution: float) -> dict:
        """Evaluate whether to trade a market."""
        research = await self.research_market(question, market_price)

        probability = research['probability']
        edge = abs(probability - market_price)

        # Confidence-adjusted edge
        confidence_multiplier = {
            'high': 1.0,
            'medium': 0.5,
            'low': 0.25
        }.get(research['confidence'], 0.5)

        adjusted_edge = edge * confidence_multiplier

        # Liquidity check
        liquidity_ok = volume > 1000  # Minimum daily volume

        # Time value
        annualized_return = (adjusted_edge / market_price) * (365 / max(time_to_resolution, 1))

        decision = {
            'question': question,
            'market_price': market_price,
            'our_probability': probability,
            'raw_edge': edge,
            'adjusted_edge': adjusted_edge,
            'annualized_return': annualized_return,
            'liquidity_adequate': liquidity_ok,
            'should_trade': adjusted_edge > 0.05 and liquidity_ok and annualized_return > 0.20,
            'direction': 'yes' if probability > market_price else 'no',
            'research': research
        }

        if decision['should_trade']:
            sizing = self.trader.size_position(
                market_price, probability,
                decision['direction']
            )
            decision['position_size'] = sizing

        return decision

    async def monitor_positions(self) -> list:
        """Check existing positions and decide whether to adjust."""
        adjustments = []

        for contract_id, position in self.trader.positions.items():
            market_info = self.active_markets.get(contract_id, {})
            current_price = market_info.get('current_price', 0.5)

            # Re-research
            research = await self.research_market(
                market_info.get('question', ''), current_price
            )

            new_probability = research['probability']

            # Check if edge has disappeared
            if position['yes_shares'] > 0:
                edge = new_probability - current_price
                if edge < 0:
                    adjustments.append({
                        'contract': contract_id,
                        'action': 'close_position',
                        'reason': 'Edge has disappeared or reversed',
                        'current_price': current_price,
                        'our_probability': new_probability
                    })
            elif position['no_shares'] > 0:
                edge = current_price - new_probability
                if edge < 0:
                    adjustments.append({
                        'contract': contract_id,
                        'action': 'close_position',
                        'reason': 'Edge has disappeared or reversed',
                        'current_price': current_price,
                        'our_probability': new_probability
                    })

        return adjustments

Portfolio Construction for Binary Outcomes

class PredictionPortfolio:
    """
    Construct and manage a portfolio of prediction market positions.
    Key difference from stocks: all positions resolve to 0 or 1.
    """

    def __init__(self, total_capital: float):
        self.capital = total_capital
        self.positions = []

    def add_position(self, contract_id: str, question: str,
                     probability: float, market_price: float,
                     direction: str, size: float,
                     correlation_group: str = 'independent'):
        self.positions.append({
            'contract_id': contract_id,
            'question': question,
            'probability': probability,
            'market_price': market_price,
            'direction': direction,
            'size': size,
            'correlation_group': correlation_group,
            'expected_pnl': self._expected_pnl(probability, market_price, direction, size),
            'max_loss': size
        })

    def _expected_pnl(self, prob, price, direction, size):
        if direction == 'yes':
            return size * (prob * (1 - price) / price - (1 - prob))
        else:
            return size * ((1 - prob) * price / (1 - price) - prob)

    def portfolio_analytics(self) -> dict:
        """Compute portfolio-level metrics."""
        total_invested = sum(p['size'] for p in self.positions)
        total_expected_pnl = sum(p['expected_pnl'] for p in self.positions)
        max_possible_loss = sum(p['max_loss'] for p in self.positions)

        # Diversification check
        by_group = {}
        for p in self.positions:
            group = p['correlation_group']
            if group not in by_group:
                by_group[group] = 0
            by_group[group] += p['size']

        concentration = max(by_group.values()) / total_invested if total_invested > 0 else 0

        # Monte Carlo portfolio simulation
        simulated_returns = self._simulate_portfolio(n_sims=10000)

        return {
            'n_positions': len(self.positions),
            'total_invested': total_invested,
            'capital_utilization': total_invested / self.capital,
            'expected_return': total_expected_pnl,
            'expected_return_pct': total_expected_pnl / total_invested * 100 if total_invested > 0 else 0,
            'max_possible_loss': max_possible_loss,
            'correlation_groups': len(by_group),
            'concentration_risk': concentration,
            'simulated_var_95': np.percentile(simulated_returns, 5),
            'simulated_median_return': np.median(simulated_returns),
            'probability_of_loss': np.mean(simulated_returns < 0)
        }

    def _simulate_portfolio(self, n_sims: int = 10000) -> np.ndarray:
        """Monte Carlo simulation of portfolio outcomes."""
        returns = np.zeros(n_sims)

        for sim in range(n_sims):
            sim_pnl = 0
            for pos in self.positions:
                # Each contract resolves randomly based on our estimated probability
                resolved_yes = np.random.random() < pos['probability']

                if pos['direction'] == 'yes':
                    if resolved_yes:
                        sim_pnl += pos['size'] * (1 / pos['market_price'] - 1)
                    else:
                        sim_pnl -= pos['size']
                else:
                    if not resolved_yes:
                        sim_pnl += pos['size'] * (1 / (1 - pos['market_price']) - 1)
                    else:
                        sim_pnl -= pos['size']

            returns[sim] = sim_pnl

        return returns

    def optimize_allocation(self, candidates: list, max_per_market: float = 0.05) -> list:
        """
        Optimize capital allocation across candidate markets.
        Maximize expected return subject to diversification constraints.
        """
        # Rank by expected return per unit risk
        scored = []
        for c in candidates:
            edge = abs(c['probability'] - c['market_price'])
            if edge <= 0.03:
                continue

            # Return per unit of maximum loss
            direction = 'yes' if c['probability'] > c['market_price'] else 'no'
            cost = c['market_price'] if direction == 'yes' else 1 - c['market_price']
            expected_return = edge / cost

            scored.append({
                **c,
                'direction': direction,
                'edge': edge,
                'expected_return': expected_return,
                'cost_per_share': cost
            })

        scored.sort(key=lambda x: -x['expected_return'])

        # Allocate capital
        allocated = []
        remaining = self.capital
        group_allocation = {}

        for candidate in scored:
            max_allocation = min(
                remaining * max_per_market,
                remaining * 0.30 if candidate['correlation_group'] not in group_allocation else
                max(0, self.capital * 0.20 - group_allocation.get(candidate['correlation_group'], 0))
            )

            if max_allocation < 10:  # Minimum trade size
                continue

            allocation = min(max_allocation, remaining * 0.10)
            allocated.append({**candidate, 'allocation': allocation})
            remaining -= allocation

            group = candidate['correlation_group']
            group_allocation[group] = group_allocation.get(group, 0) + allocation

            if remaining < self.capital * 0.1:
                break

        return allocated

Risk Management

class PredictionMarketRiskManager:
    """Risk management specific to prediction markets."""

    def __init__(self, max_drawdown_pct: float = 20,
                 max_single_position_pct: float = 5):
        self.max_drawdown_pct = max_drawdown_pct
        self.max_single_position_pct = max_single_position_pct
        self.peak_bankroll = 0
        self.current_bankroll = 0

    def pre_trade_check(self, trade: dict, portfolio: PredictionPortfolio) -> dict:
        """Run pre-trade risk checks."""
        checks = {
            'position_size_ok': trade['size'] <= portfolio.capital * self.max_single_position_pct / 100,
            'capital_available': trade['size'] <= portfolio.capital * 0.90,  # Keep 10% reserve
            'drawdown_ok': self._check_drawdown(),
            'concentration_ok': self._check_concentration(trade, portfolio),
            'edge_sufficient': trade.get('edge', 0) > 0.03
        }

        checks['all_passed'] = all(checks.values())

        if not checks['all_passed']:
            failed = [k for k, v in checks.items() if not v and k != 'all_passed']
            checks['failed_checks'] = failed

        return checks

    def _check_drawdown(self) -> bool:
        if self.peak_bankroll == 0:
            return True
        drawdown = (self.peak_bankroll - self.current_bankroll) / self.peak_bankroll * 100
        return drawdown < self.max_drawdown_pct

    def _check_concentration(self, trade: dict, portfolio: PredictionPortfolio) -> bool:
        group = trade.get('correlation_group', 'independent')
        group_exposure = sum(
            p['size'] for p in portfolio.positions
            if p['correlation_group'] == group
        )
        return (group_exposure + trade['size']) < portfolio.capital * 0.25

Key Takeaways

1. Expected value in prediction markets is simply your_probability minus market_price; trade only when this edge exceeds your transaction costs and minimum threshold
2. Kelly Criterion gives optimal sizing but is very aggressive; use half-Kelly or quarter-Kelly in practice to survive variance
3. Cross-platform arbitrage requires prices to sum to less than 1 after fees; these opportunities are rare but risk-free when found
4. Market making in prediction markets earns the spread while managing inventory; skew quotes to flatten position when inventory grows
5. AI trading agents combine LLM research with systematic position sizing; the Olas/Polystrat approach automates the full pipeline from research to execution
6. Prediction market portfolios need Monte Carlo simulation because all positions are binary (no partial outcomes)
7. Correlation management is critical: political markets are often highly correlated, and a single wrong assessment can wipe out a concentrated portfolio
8. Pre-trade risk checks (position limits, drawdown limits, concentration limits, minimum edge) prevent the most common causes of ruin