AgentMarket

Interface for accessing prediction market data including questions, probabilities, trading, and positions

Overview

AgentMarket is the core interface for interacting with prediction markets across different platforms. It provides a unified API for accessing market metadata, current probabilities, liquidity, trading operations, and position management.

Import

from prediction_market_agent_tooling.markets.agent_market import AgentMarket

Basic Properties

Market Identity

id · str · required

Unique identifier for the market

url · str · required

Direct URL to view the market on the platform

question · str · required

The market’s question or title

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    logger.info(f"Processing market: {market.question}")
    logger.info(f"Market URL: {market.url}")
    logger.info(f"Market ID: {market.id}")

Market Outcomes

outcomes · list[str] · required

List of possible outcomes (e.g., [“Yes”, “No”] for binary markets)

probabilities · dict[str, Probability] · required

Current probability for each outcome

def analyze_market(self, market: AgentMarket) -> None:
    # Access outcomes and probabilities
    for outcome in market.outcomes:
        prob = market.probabilities[outcome]
        print(f"{outcome}: {prob:.2%}")
    
    # For binary markets
    if len(market.outcomes) == 2:
        yes_prob = market.probabilities["Yes"]
        no_prob = market.probabilities["No"]

Market Type

market_type · MarketType · required

Platform where the market exists (OMEN, MANIFOLD, etc.)

from prediction_market_agent_tooling.markets.markets import MarketType
from prediction_market_agent_tooling.markets.omen.omen import OmenAgentMarket

def get_betting_strategy(self, market: AgentMarket) -> BettingStrategy:
    if isinstance(market, OmenAgentMarket):
        # Use Kelly criterion for Omen
        return FullBinaryKellyBettingStrategy(
            max_position_amount=USD(5),
            max_price_impact=0.7
        )
    else:
        # Use conservative strategy for other platforms
        return super().get_betting_strategy(market)

Time Properties

created_time · datetime · required

When the market was created

close_time · datetime | None

When the market will close for trading

finalized_time · datetime | None

When the market was finalized/resolved

from datetime import timedelta
from prediction_market_agent_tooling.tools.utils import utcnow, check_not_none

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    # Skip markets closing very soon
    if market.close_time:
        time_until_close = check_not_none(market.close_time) - utcnow()
        if time_until_close < timedelta(hours=1):
            logger.info(f"Market closes too soon: {market.url}")
            return False
    
    # Only trade on recent markets
    market_age = utcnow() - market.created_time
    if market_age > timedelta(days=30):
        return False
    
    return True

Scalar Market Properties

For numeric prediction markets:

lower_bound · float | None

Minimum possible value

upper_bound · float | None

Maximum possible value

def answer_scalar_market(
    self, 
    market: AgentMarket
) -> ScalarProbabilisticAnswer | None:
    if market.upper_bound is None or market.lower_bound is None:
        raise ValueError("Market upper and lower bounds must be set")
    
    logger.info(
        f"Scalar market range: {market.lower_bound} to {market.upper_bound}"
    )
    
    prediction = self.agent.predict_scalar(
        market.question,
        market.upper_bound,
        market.lower_bound
    )
    return prediction.outcome_prediction

Market Statistics

get_liquidity

Get the current liquidity available in the market.

def get_liquidity(self) -> TokenAmount:
    pass

return · TokenAmount

Current liquidity in the market’s native token

from prediction_market_agent_tooling.gtypes import USD

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    # Require minimum liquidity
    min_liquidity = market.get_in_token(USD(5))
    
    if market.get_liquidity() < min_liquidity:
        logger.info(
            f"Skipping market with low liquidity: {market.url}"
        )
        return False
    
    return True

volume

volume · TokenAmount | None

Total trading volume in the market

def analyze_market_activity(self, market: AgentMarket) -> None:
    if market.volume:
        logger.info(f"Market volume: {market.volume}")
        logger.info(f"Market liquidity: {market.get_liquidity()}")

Trading Operations

get_trade_balance

Get available balance for trading on this market.

def get_trade_balance(self, api_keys: APIKeys) -> TokenAmount:
    pass

api_keys · APIKeys · required

API keys for accessing wallet/account

return · TokenAmount

Available balance in the market’s token

from prediction_market_agent.utils import APIKeys

def get_betting_strategy(self, market: AgentMarket) -> BettingStrategy:
    trading_balance = market.get_trade_balance(APIKeys())
    
    max_bet = get_maximum_possible_bet_amount(
        min_=USD(1),
        max_=USD(5),
        trading_balance=trading_balance
    )
    
    return FullBinaryKellyBettingStrategy(
        max_position_amount=max_bet,
        max_price_impact=0.7
    )

place_bet

Execute a bet on the market.

def place_bet(
    self,
    outcome: str,
    amount: TokenAmount,
    api_keys: APIKeys
) -> None:
    pass

outcome · str · required

The outcome to bet on (must be in market.outcomes)

amount · TokenAmount · required

Amount to bet in the market’s token

api_keys · APIKeys · required

API keys for executing the transaction

# Note: Usually you don't call this directly - the framework handles it
# But here's how it works internally:

market.place_bet(
    outcome="Yes",
    amount=market.get_in_token(USD(2)),
    api_keys=APIKeys()
)

Position Management

get_positions

Get your current positions in this market.

def get_positions(self, user_id: str) -> list[Position]:
    pass

user_id · str · required

User identifier (address or account ID)

return · list[Position]

List of current positions in the market

from prediction_market_agent_tooling.markets.data_models import Position

def analyze_existing_positions(self, market: AgentMarket) -> None:
    user_id = market.get_user_id(api_keys=APIKeys())
    positions = market.get_positions(user_id)
    
    for position in positions:
        logger.info(
            f"Outcome: {position.outcome}, "
            f"Amount: {position.amounts[position.outcome]}"
        )

get_most_recent_trade_datetime

Get the timestamp of the most recent trade by a user.

def get_most_recent_trade_datetime(self, user_id: str) -> datetime | None:
    pass

user_id · str · required

User identifier

return · datetime | None

Timestamp of most recent trade, or None if no trades

from prediction_market_agent_tooling.tools.utils import utcnow

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    # Check if we've traded recently on this market
    user_id = market.get_user_id(api_keys=APIKeys())
    last_trade = market.get_most_recent_trade_datetime(user_id=user_id)
    
    if last_trade is None:
        return True  # Never traded, OK to trade
    
    # Skip if we traded within last 24 hours
    time_since_trade = utcnow() - last_trade
    if time_since_trade < timedelta(days=1):
        logger.info(f"Recently traded on {market.url}, skipping")
        return False
    
    return True

get_user_id

Get the user identifier for the current agent.

def get_user_id(self, api_keys: APIKeys) -> str:
    pass

api_keys · APIKeys · required

API keys to derive user ID from

return · str

User identifier (wallet address or account ID)

def check_existing_position(self, market: AgentMarket) -> bool:
    user_id = market.get_user_id(api_keys=APIKeys())
    positions = market.get_positions(user_id)
    return len(positions) > 0

Resolution Status

resolution · str | None

The resolved outcome if market is finalized, None otherwise

def analyze_resolved_markets(self, markets: list[AgentMarket]) -> None:
    resolved = [m for m in markets if m.resolution is not None]
    
    for market in resolved:
        logger.info(
            f"Market '{market.question}' resolved to: {market.resolution}"
        )

Token Conversion

get_in_token

Convert an amount to the market’s native token.

def get_in_token(self, amount: TokenAmount) -> TokenAmount:
    pass

amount · TokenAmount · required

Amount to convert (e.g., USD)

return · TokenAmount

Equivalent amount in market’s token

from prediction_market_agent_tooling.gtypes import USD

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    # Require at least $10 liquidity
    min_liquidity_usd = USD(10)
    min_liquidity_native = market.get_in_token(min_liquidity_usd)
    
    if market.get_liquidity() < min_liquidity_native:
        return False
    
    return True

Fetching Markets

get_binary_markets

Fetch binary (yes/no) markets from a platform.

from prediction_market_agent_tooling.markets.markets import (
    get_binary_markets,
    MarketType,
    FilterBy,
    SortBy
)

markets = get_binary_markets(
    limit=50,
    market_type=MarketType.OMEN,
    filter_by=FilterBy.OPEN,
    sort_by=SortBy.NEWEST
)

limit · int · required

Maximum number of markets to fetch

market_type · MarketType · required

Platform to fetch from

filter_by · FilterBy · default: "FilterBy.OPEN"

Filter criteria (OPEN, RESOLVED, etc.)

sort_by · SortBy · default: "SortBy.NONE"

Sort order (NEWEST, HIGHEST_LIQUIDITY, etc.)

Filter and Sort Options

FilterBy

from prediction_market_agent_tooling.markets.agent_market import FilterBy

FilterBy.OPEN          # Only open markets
FilterBy.RESOLVED      # Only resolved markets  
FilterBy.NONE          # No filtering

SortBy

from prediction_market_agent_tooling.markets.agent_market import SortBy

SortBy.NEWEST              # Newest markets first
SortBy.HIGHEST_LIQUIDITY   # Highest liquidity first
SortBy.CLOSING_SOONEST     # Closing soonest first
SortBy.NONE                # No specific order

Platform-Specific Implementations

OmenAgentMarket

from prediction_market_agent_tooling.markets.omen.omen import OmenAgentMarket

if isinstance(market, OmenAgentMarket):
    # Access Omen-specific properties
    logger.info(f"Condition ID: {market.condition.id}")
    logger.info(f"Collateral token: {market.collateral_token_contract_address_checksummed}")

Complete Example

from prediction_market_agent_tooling.deploy.agent import DeployableTraderAgent
from prediction_market_agent_tooling.markets.agent_market import (
    AgentMarket,
    FilterBy,
    SortBy
)
from prediction_market_agent_tooling.markets.markets import MarketType
from prediction_market_agent_tooling.markets.data_models import ProbabilisticAnswer
from prediction_market_agent_tooling.gtypes import USD, Probability
from prediction_market_agent_tooling.tools.utils import utcnow
from datetime import timedelta

class MyMarketAgent(DeployableTraderAgent):
    get_markets_sort_by = SortBy.HIGHEST_LIQUIDITY
    bet_on_n_markets_per_run = 5
    
    def verify_market(
        self, 
        market_type: MarketType, 
        market: AgentMarket
    ) -> bool:
        # Check liquidity
        if market.get_liquidity() < market.get_in_token(USD(10)):
            logger.info(f"Low liquidity: {market.url}")
            return False
        
        # Check time until close
        if market.close_time:
            time_until_close = market.close_time - utcnow()
            if time_until_close < timedelta(hours=2):
                logger.info(f"Closing too soon: {market.url}")
                return False
        
        # Check if already traded recently
        user_id = market.get_user_id(api_keys=APIKeys())
        last_trade = market.get_most_recent_trade_datetime(user_id)
        if last_trade and (utcnow() - last_trade) < timedelta(days=1):
            logger.info(f"Recently traded: {market.url}")
            return False
        
        return True
    
    def answer_binary_market(
        self, 
        market: AgentMarket
    ) -> ProbabilisticAnswer | None:
        # Log market details
        logger.info(f"Analyzing: {market.question}")
        logger.info(f"Current probabilities: {market.probabilities}")
        logger.info(f"Liquidity: {market.get_liquidity()}")
        
        # Your prediction logic
        prediction = analyze_market(market)
        
        return ProbabilisticAnswer(
            p_yes=prediction.probability,
            confidence=prediction.confidence,
            reasoning=prediction.reasoning
        )

See Also

• DeployableTraderAgent - Base class for building agents
• ProbabilisticAnswer - Prediction data models
• Market Filtering Guide - Advanced market selection