DeployableTraderAgent

Base class for creating automated prediction market trading agents with built-in market processing, betting strategies, and trade execution

Overview

DeployableTraderAgent is the core abstract base class from prediction-market-agent-tooling for building automated trading agents. It provides a complete framework for discovering markets, generating predictions, managing betting strategies, and executing trades across multiple prediction market platforms.

Import

from prediction_market_agent_tooling.deploy.agent import DeployableTraderAgent

Basic Usage

from prediction_market_agent_tooling.deploy.agent import DeployableTraderAgent
from prediction_market_agent_tooling.markets.agent_market import AgentMarket
from prediction_market_agent_tooling.markets.data_models import ProbabilisticAnswer
from prediction_market_agent_tooling.gtypes import Probability

class MyTradingAgent(DeployableTraderAgent):
    bet_on_n_markets_per_run = 2
    
    def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
        # Your prediction logic here
        return ProbabilisticAnswer(
            p_yes=Probability(0.65),
            confidence=0.8,
            reasoning="Analysis suggests 65% probability"
        )

Core Methods

answer_binary_market

Generate a prediction for a binary (yes/no) market.

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    pass

market · AgentMarket · required

The market to analyze and predict

return · ProbabilisticAnswer | None

Your prediction with probability, confidence, and reasoning. Return None to skip the market.

Example:

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    # Run your analysis
    prediction = self.agent.predict(market.question)
    
    return ProbabilisticAnswer(
        p_yes=prediction.probability,
        confidence=prediction.confidence,
        reasoning=prediction.reasoning
    )

answer_categorical_market

Generate a prediction for a categorical (multiple choice) market.

def answer_categorical_market(
    self, 
    market: AgentMarket
) -> CategoricalProbabilisticAnswer | None:
    pass

market · AgentMarket · required

The categorical market to analyze

return · CategoricalProbabilisticAnswer | None

Probabilities for each outcome. Return None to skip the market.

Example:

def answer_categorical_market(
    self, 
    market: AgentMarket
) -> CategoricalProbabilisticAnswer | None:
    prediction = self.agent.predict_categorical(
        market.question, 
        market.outcomes
    )
    return prediction.outcome_prediction

answer_scalar_market

Generate a prediction for a scalar (numeric range) market.

def answer_scalar_market(
    self, 
    market: AgentMarket
) -> ScalarProbabilisticAnswer | None:
    pass

market · AgentMarket · required

The scalar market with upper and lower bounds

return · ScalarProbabilisticAnswer | None

Prediction within the market’s defined range. Return None to skip.

Example:

def answer_scalar_market(
    self, 
    market: AgentMarket
) -> ScalarProbabilisticAnswer | None:
    if market.upper_bound is None or market.lower_bound is None:
        raise ValueError("Market bounds must be set")
    
    prediction = self.agent.predict_scalar(
        market.question,
        market.upper_bound,
        market.lower_bound
    )
    return prediction.outcome_prediction

verify_market

Filter markets before processing. Override to implement custom market selection logic.

def verify_market(
    self, 
    market_type: MarketType, 
    market: AgentMarket
) -> bool:
    pass

market_type · MarketType · required

The type of market platform (OMEN, MANIFOLD, etc.)

market · AgentMarket · required

The market to verify

return · bool

True to process the market, False to skip it

Example:

def verify_market(
    self, 
    market_type: MarketType, 
    market: AgentMarket
) -> bool:
    # Only process markets with sufficient liquidity
    if market.get_liquidity() < market.get_in_token(USD(5)):
        logger.info(f"Skipping market with low liquidity: {market.url}")
        return False
    
    # Skip saturated markets
    if market_is_saturated(market):
        logger.info(f"Skipping saturated market: {market.url}")
        return False
    
    return True

get_betting_strategy

Customize betting behavior for each market.

def get_betting_strategy(self, market: AgentMarket) -> BettingStrategy:
    pass

market · AgentMarket · required

The market being traded

return · BettingStrategy

Strategy controlling bet sizing and risk parameters

Example:

from prediction_market_agent_tooling.deploy.betting_strategy import (
    FullBinaryKellyBettingStrategy
)

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

load

Initialize agent resources before market processing begins.

def load(self) -> None:
    pass

Example:

def load(self) -> None:
    super().load()
    
    # Initialize your prediction model
    self.agent = PredictionProphetAgent(
        research_agent=Agent(OpenAIModel("gpt-4o")),
        prediction_agent=Agent(OpenAIModel("gpt-4o")),
        logger=logger
    )
    
    # Set up any required caches or handlers
    self.pinecone_handler = PineconeHandler()

before_process_markets

Run setup tasks before market processing starts.

def before_process_markets(self, market_type: MarketType) -> None:
    pass

market_type · MarketType · required

The market platform being processed

Example:

def before_process_markets(self, market_type: MarketType) -> None:
    # Update vector database with latest markets
    self.pinecone_handler.insert_all_omen_markets_if_not_exists()
    super().before_process_markets(market_type=market_type)

get_markets

Retrieve and filter markets to process. Override for custom market selection.

def get_markets(
    self, 
    market_type: MarketType
) -> Sequence[AgentMarket]:
    pass

market_type · MarketType · required

The market platform to fetch from

return · Sequence[AgentMarket]

List of markets to process

Example:

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

def get_markets(
    self, 
    market_type: MarketType
) -> Sequence[AgentMarket]:
    # Only process markets closing within 14 days
    max_close_time = utcnow() + timedelta(days=14)
    
    markets = super().get_markets(market_type)
    filtered_markets = [
        m for m in markets 
        if check_not_none(m.close_time) < max_close_time
    ]
    
    return filtered_markets

Configuration Attributes

bet_on_n_markets_per_run

bet_on_n_markets_per_run · int · default: "1"

Maximum number of markets to trade on per execution

class MyAgent(DeployableTraderAgent):
    bet_on_n_markets_per_run = 4  # Trade on up to 4 markets per run

n_markets_to_fetch

n_markets_to_fetch · int · default: "50"

Number of markets to fetch from the platform

class SkewAgent(DeployableTraderAgent):
    n_markets_to_fetch = 1000  # Fetch many markets for filtering

get_markets_sort_by

get_markets_sort_by · SortBy · default: "SortBy.NONE"

Sort order for fetched markets

from prediction_market_agent_tooling.markets.agent_market import SortBy

class NewMarketAgent(DeployableTraderAgent):
    get_markets_sort_by = SortBy.NEWEST

same_market_trade_interval

same_market_trade_interval · TradeInterval · default: "Never"

Time interval between trades on the same market

from datetime import timedelta
from prediction_market_agent_tooling.deploy.trade_interval import FixedInterval

class MyAgent(DeployableTraderAgent):
    same_market_trade_interval = FixedInterval(timedelta(days=7))

supported_markets

supported_markets · list[MarketType] · default: "All markets"

List of supported market platforms

from prediction_market_agent_tooling.markets.markets import MarketType

class OmenOnlyAgent(DeployableTraderAgent):
    supported_markets = [MarketType.OMEN]

trade_on_markets_created_after

trade_on_markets_created_after · DatetimeUTC | None · default: "None"

Only trade on markets created after this timestamp

from prediction_market_agent_tooling.tools.utils import DatetimeUTC

class RecentMarketsAgent(DeployableTraderAgent):
    trade_on_markets_created_after = DatetimeUTC(2024, 10, 31, 0)

Deployment

run

Execute the agent’s main trading loop.

def run(self, market_type: MarketType) -> None:
    pass

market_type · MarketType · required

The market platform to trade on

Example:

from prediction_market_agent_tooling.markets.markets import MarketType

agent = MyTradingAgent()
agent.run(market_type=MarketType.OMEN)

deploy_local

Run the agent locally with continuous execution.

def deploy_local(
    self,
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None = None
) -> None:
    pass

market_type · MarketType · required

The market platform to trade on

sleep_time · float · required

Seconds to sleep between runs

run_time · float | None · default: "None"

Total runtime in seconds (None for infinite)

Example:

if __name__ == "__main__":
    agent = MyTradingAgent(
        place_trades=True,
        store_predictions=True
    )
    agent.deploy_local(
        market_type=MarketType.OMEN,
        sleep_time=300,  # Run every 5 minutes
        run_time=3600    # Run for 1 hour
    )

Constructor Parameters

DeployableTraderAgent(
    place_trades: bool = True,
    store_predictions: bool = True,
    store_trades: bool = True,
    enable_langfuse: bool = True
)

place_trades · bool · default: "True"

Whether to actually execute trades (False for dry-run mode)

store_predictions · bool · default: "True"

Whether to store predictions to database

store_trades · bool · default: "True"

Whether to store trade records to database

enable_langfuse · bool · default: "True"

Enable Langfuse observability tracking

Complete Example

from prediction_market_agent_tooling.deploy.agent import DeployableTraderAgent
from prediction_market_agent_tooling.deploy.betting_strategy import (
    BettingStrategy,
    FullBinaryKellyBettingStrategy
)
from prediction_market_agent_tooling.markets.agent_market import AgentMarket
from prediction_market_agent_tooling.markets.data_models import ProbabilisticAnswer
from prediction_market_agent_tooling.markets.markets import MarketType
from prediction_market_agent_tooling.gtypes import USD, Probability
from prediction_prophet.benchmark.agents import PredictionProphetAgent

class MyPredictionAgent(DeployableTraderAgent):
    bet_on_n_markets_per_run = 2
    
    def load(self) -> None:
        super().load()
        # Initialize your prediction model
        self.agent = PredictionProphetAgent(
            research_agent=Agent(OpenAIModel("gpt-4o")),
            prediction_agent=Agent(OpenAIModel("gpt-4o")),
            logger=logger
        )
    
    def verify_market(
        self, 
        market_type: MarketType, 
        market: AgentMarket
    ) -> bool:
        # Only process markets with sufficient liquidity
        if market.get_liquidity() < market.get_in_token(USD(10)):
            return False
        return True
    
    def answer_binary_market(
        self, 
        market: AgentMarket
    ) -> ProbabilisticAnswer | None:
        prediction = self.agent.predict(market.question)
        
        if prediction.outcome_prediction is None:
            return None
            
        return prediction.outcome_prediction.to_probabilistic_answer()
    
    def get_betting_strategy(
        self, 
        market: AgentMarket
    ) -> BettingStrategy:
        return FullBinaryKellyBettingStrategy(
            max_position_amount=USD(5),
            max_price_impact=0.7
        )

if __name__ == "__main__":
    agent = MyPredictionAgent()
    agent.deploy_local(
        market_type=MarketType.OMEN,
        sleep_time=300
    )

See Also

• AgentMarket - Market interface for accessing market data
• ProbabilisticAnswer - Data models for predictions
• Betting Strategies - Configure bet sizing and risk