Skip to main content

​
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

Build docs developers (and LLMs) love

Get started for freeTalk to us