Deploy Module

prediction_market_agent_tooling.deploy

agent

MAX_AVAILABLE_MARKETS module-attribute

MAX_AVAILABLE_MARKETS = 1000

AnsweredEnum

Bases: str, Enum

ANSWERED class-attribute instance-attribute

ANSWERED = 'answered'

NOT_ANSWERED class-attribute instance-attribute

NOT_ANSWERED = 'not_answered'

AgentTagEnum

Bases: str, Enum

PREDICTOR class-attribute instance-attribute

PREDICTOR = 'predictor'

TRADER class-attribute instance-attribute

TRADER = 'trader'

DeployableAgent

DeployableAgent(
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
)

Subclass this class to create agent with standardized interface.

Source code in prediction_market_agent_tooling/deploy/agent.py

def __init__(
    self,
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
) -> None:
    self.start_time = utcnow()
    self.enable_langfuse = enable_langfuse
    self.api_keys = APIKeys()
    self.initialize_langfuse()
    self.load()

start_time instance-attribute

start_time = utcnow()

enable_langfuse instance-attribute

enable_langfuse = enable_langfuse

api_keys instance-attribute

api_keys = APIKeys()

session_id cached property

session_id: str

initialize_langfuse

initialize_langfuse() -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def initialize_langfuse(self) -> None:
    initialize_langfuse(self.enable_langfuse)

langfuse_update_current_trace

langfuse_update_current_trace(
    name: str | None = None,
    input: Any | None = None,
    output: Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None

Provide some useful default arguments when updating the current trace in our agents.

Source code in prediction_market_agent_tooling/deploy/agent.py

def langfuse_update_current_trace(
    self,
    name: str | None = None,
    input: t.Any | None = None,
    output: t.Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: t.Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None:
    """
    Provide some useful default arguments when updating the current trace in our agents.
    """
    langfuse_context.update_current_trace(
        name=name,
        input=input,
        output=output,
        user_id=user_id or getpass.getuser(),
        session_id=session_id or self.session_id,
        # All traces within a single run execution will be grouped under a single session.
        version=version or APIKeys().LANGFUSE_DEPLOYMENT_VERSION,
        # Optionally, mark the current deployment with version (e.g. add git commit hash during docker building).
        release=release,
        metadata=metadata,
        tags=tags,
        public=public,
    )

load

load() -> None

Implement this method to load arbitrary instances needed across the whole run of the agent.

Do not customize init method.

Source code in prediction_market_agent_tooling/deploy/agent.py

def load(self) -> None:
    """
    Implement this method to load arbitrary instances needed across the whole run of the agent.

    Do not customize __init__ method.
    """

deploy_local

deploy_local(
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None

Run the agent in the forever cycle every sleep_time seconds, until the run_time is met.

Source code in prediction_market_agent_tooling/deploy/agent.py

def deploy_local(
    self,
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None:
    """
    Run the agent in the forever cycle every `sleep_time` seconds, until the `run_time` is met.
    """
    start_time = time.time()
    while run_time is None or time.time() - start_time < run_time:
        self.run(market_type=market_type)
        time.sleep(sleep_time)

run

run(market_type: MarketType) -> None

Run single iteration of the agent.

Source code in prediction_market_agent_tooling/deploy/agent.py

def run(self, market_type: MarketType) -> None:
    """
    Run single iteration of the agent.
    """
    raise NotImplementedError("This method must be implemented by the subclass.")

get_gcloud_fname

get_gcloud_fname(market_type: MarketType) -> str

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_gcloud_fname(self, market_type: MarketType) -> str:
    return f"{self.__class__.__name__.lower()}-{market_type}-{utcnow().strftime('%Y-%m-%d--%H-%M-%S')}"

DeployablePredictionAgent

DeployablePredictionAgent(
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
)

Bases: DeployableAgent

Subclass this class to create your own prediction market agent.

The agent will process markets and make predictions.

Source code in prediction_market_agent_tooling/deploy/agent.py

def __init__(
    self,
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
) -> None:
    super().__init__(enable_langfuse=enable_langfuse)
    self.store_predictions = store_predictions

AGENT_TAG class-attribute instance-attribute

AGENT_TAG: AgentTagEnum = PREDICTOR

bet_on_n_markets_per_run class-attribute instance-attribute

bet_on_n_markets_per_run: int = 1

n_markets_to_fetch class-attribute instance-attribute

n_markets_to_fetch: int = MAX_AVAILABLE_MARKETS

trade_on_markets_created_after class-attribute instance-attribute

trade_on_markets_created_after: DatetimeUTC | None = None

get_markets_sort_by class-attribute instance-attribute

get_markets_sort_by: SortBy = CLOSING_SOONEST

get_markets_filter_by class-attribute instance-attribute

get_markets_filter_by: FilterBy = OPEN

allow_invalid_questions class-attribute instance-attribute

allow_invalid_questions: bool = False

same_market_trade_interval class-attribute instance-attribute

same_market_trade_interval: TradeInterval = FixedInterval(
    timedelta(hours=24)
)

min_balance_to_keep_in_native_currency class-attribute instance-attribute

min_balance_to_keep_in_native_currency: xDai | None = (
    MINIMUM_NATIVE_TOKEN_IN_EOA_FOR_FEES
)

supported_markets class-attribute instance-attribute

supported_markets: Sequence[MarketType] = [METACULUS]

store_predictions instance-attribute

store_predictions = store_predictions

agent_name property

agent_name: str

fetch_categorical_markets property

fetch_categorical_markets: bool

start_time instance-attribute

start_time = utcnow()

enable_langfuse instance-attribute

enable_langfuse = enable_langfuse

api_keys instance-attribute

api_keys = APIKeys()

session_id cached property

session_id: str

initialize_langfuse

initialize_langfuse() -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def initialize_langfuse(self) -> None:
    super().initialize_langfuse()
    # Auto-observe all the methods where it makes sense, so that subclassses don't need to do it manually.
    self.verify_market = observe()(self.verify_market)  # type: ignore[method-assign]
    self.answer_binary_market = observe()(self.answer_binary_market)  # type: ignore[method-assign]
    self.answer_categorical_market = observe()(self.answer_categorical_market)  # type: ignore[method-assign]
    self.process_market = observe()(self.process_market)  # type: ignore[method-assign]

update_langfuse_trace_by_market

update_langfuse_trace_by_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    self.langfuse_update_current_trace(
        # UI allows to do filtering by these.
        metadata={
            "agent_class": self.__class__.__name__,
            "market_id": market.id,
            "market_question": market.question,
            "market_outcomes": market.outcomes,
        },
    )

update_langfuse_trace_by_processed_market

update_langfuse_trace_by_processed_market(
    market_type: MarketType,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_processed_market(
    self, market_type: MarketType, processed_market: ProcessedMarket | None
) -> None:
    self.langfuse_update_current_trace(
        tags=[
            self.AGENT_TAG,
            (
                AnsweredEnum.ANSWERED
                if processed_market is not None
                else AnsweredEnum.NOT_ANSWERED
            ),
            market_type.value,
        ]
    )

check_min_required_balance_to_operate

check_min_required_balance_to_operate(
    market_type: MarketType,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_operate(self, market_type: MarketType) -> None:
    api_keys = APIKeys()

    if not market_type.market_class.verify_operational_balance(api_keys):
        raise CantPayForGasError(
            f"{api_keys=} doesn't have enough operational balance."
        )

verify_market

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

Subclasses can implement their own logic instead of this one, or on top of this one. By default, it allows only markets where user didn't bet recently and it's a reasonable question.

Source code in prediction_market_agent_tooling/deploy/agent.py

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    """
    Subclasses can implement their own logic instead of this one, or on top of this one.
    By default, it allows only markets where user didn't bet recently and it's a reasonable question.
    """
    if market.have_bet_on_market_since(
        keys=APIKeys(), since=self.same_market_trade_interval.get(market=market)
    ):
        logger.info(
            f"Market already bet on within {self.same_market_trade_interval}."
        )
        return False

    # Manifold allows to bet only on markets with probability between 1 and 99.
    if market_type == MarketType.MANIFOLD:
        probability_yes = market.probabilities[
            market.get_outcome_str_from_bool(True)
        ]
        if not probability_yes or not 1 < probability_yes < 99:
            logger.info("Manifold's market probability not in the range 1-99.")
            return False

    # Do as a last check, as it uses paid OpenAI API.
    if not is_predictable_binary(market.question):
        logger.info("Market question is not predictable.")
        return False

    if not self.allow_invalid_questions and is_invalid(market.question):
        logger.info("Market question is invalid.")
        return False

    return True

answer_categorical_market

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

Source code in prediction_market_agent_tooling/deploy/agent.py

def answer_categorical_market(
    self, market: AgentMarket
) -> CategoricalProbabilisticAnswer | None:
    raise NotImplementedError("This method must be implemented by the subclass")

answer_binary_market

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

Answer the binary market.

If this method is not overridden by the subclass, it will fall back to using answer_categorical_market(). Therefore, subclasses only need to implement answer_categorical_market() if they want to handle both types of markets.

Source code in prediction_market_agent_tooling/deploy/agent.py

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    """
    Answer the binary market.

    If this method is not overridden by the subclass, it will fall back to using
    answer_categorical_market(). Therefore, subclasses only need to implement
    answer_categorical_market() if they want to handle both types of markets.
    """
    raise NotImplementedError(
        "Either this method, or answer_categorical_market, must be implemented by the subclass."
    )

get_markets

get_markets(
    market_type: MarketType,
) -> t.Sequence[AgentMarket]

Override this method to customize what markets will fetch for processing.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_markets(
    self,
    market_type: MarketType,
) -> t.Sequence[AgentMarket]:
    """
    Override this method to customize what markets will fetch for processing.
    """
    cls = market_type.market_class
    # Fetch the soonest closing markets to choose from
    available_markets = cls.get_markets(
        limit=self.n_markets_to_fetch,
        sort_by=self.get_markets_sort_by,
        filter_by=self.get_markets_filter_by,
        created_after=self.trade_on_markets_created_after,
        fetch_categorical_markets=self.fetch_categorical_markets,
    )
    return available_markets

before_process_market

before_process_market(
    market_type: MarketType, market: AgentMarket
) -> None

Executed before processing of each market.

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    """
    Executed before processing of each market.
    """
    api_keys = APIKeys()

    if market_type.is_blockchain_market:
        # Exchange wxdai back to xdai if the balance is getting low, so we can keep paying for fees.
        if self.min_balance_to_keep_in_native_currency is not None:
            send_keeping_token_to_eoa_xdai(
                api_keys,
                min_required_balance=self.min_balance_to_keep_in_native_currency,
                multiplier=3,
            )

build_answer

build_answer(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_answer(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None:
    if verify_market and not self.verify_market(market_type, market):
        logger.info(f"Market '{market.question}' doesn't meet the criteria.")
        return None

    logger.info(f"Answering market '{market.question}'.")

    if market.is_binary:
        try:
            binary_answer = self.answer_binary_market(market)
            return (
                CategoricalProbabilisticAnswer.from_probabilistic_answer(
                    binary_answer
                )
                if binary_answer is not None
                else None
            )
        except NotImplementedError:
            logger.info(
                "answer_binary_market() not implemented, falling back to answer_categorical_market()"
            )

    return self.answer_categorical_market(market)

process_market

process_market(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedMarket | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedMarket | None:
    self.update_langfuse_trace_by_market(market_type, market)
    logger.info(
        f"Processing market {market.question=} from {market.url=} with liquidity {market.get_liquidity()}."
    )

    answer = self.build_answer(
        market=market, market_type=market_type, verify_market=verify_market
    )

    processed_market = (
        ProcessedMarket(answer=answer) if answer is not None else None
    )

    self.update_langfuse_trace_by_processed_market(market_type, processed_market)
    logger.info(
        f"Processed market {market.question=} from {market.url=} with {answer=}."
    )
    return processed_market

after_process_market

after_process_market(
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None

Executed after processing of each market.

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None:
    """
    Executed after processing of each market.
    """
    keys = APIKeys()
    if self.store_predictions:
        market.store_prediction(
            processed_market=processed_market, keys=keys, agent_name=self.agent_name
        )
    else:
        logger.info(
            f"Prediction {processed_market} not stored because {self.store_predictions=}."
        )

before_process_markets

before_process_markets(market_type: MarketType) -> None

Executed before market processing loop starts.

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_markets(self, market_type: MarketType) -> None:
    """
    Executed before market processing loop starts.
    """
    api_keys = APIKeys()
    self.check_min_required_balance_to_operate(market_type)
    market_type.market_class.redeem_winnings(api_keys)

process_markets

process_markets(market_type: MarketType) -> None

Processes bets placed by agents on a given market.

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_markets(self, market_type: MarketType) -> None:
    """
    Processes bets placed by agents on a given market.
    """
    logger.info("Start processing of markets.")
    available_markets = self.get_markets(market_type)

    logger.info(
        f"Fetched {len(available_markets)=} markets to process, going to process {self.bet_on_n_markets_per_run=}."
    )
    processed = 0

    for market_idx, market in enumerate(available_markets):
        logger.info(
            f"Going to process market {market.url}: {market_idx+1} / {len(available_markets)}."
        )
        self.before_process_market(market_type, market)
        processed_market = self.process_market(market_type, market)
        self.after_process_market(market_type, market, processed_market)

        if processed_market is not None:
            processed += 1

        if processed == self.bet_on_n_markets_per_run:
            break

    logger.info(
        f"All markets processed. Successfully processed {processed}/{len(available_markets)}."
    )

after_process_markets

after_process_markets(market_type: MarketType) -> None

Executed after market processing loop ends.

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_markets(self, market_type: MarketType) -> None:
    """
    Executed after market processing loop ends.
    """

run

run(market_type: MarketType) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def run(self, market_type: MarketType) -> None:
    if market_type not in self.supported_markets:
        raise ValueError(
            f"Only {self.supported_markets} are supported by this agent."
        )
    self.before_process_markets(market_type)
    self.process_markets(market_type)
    self.after_process_markets(market_type)

langfuse_update_current_trace

langfuse_update_current_trace(
    name: str | None = None,
    input: Any | None = None,
    output: Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None

Provide some useful default arguments when updating the current trace in our agents.

Source code in prediction_market_agent_tooling/deploy/agent.py

def langfuse_update_current_trace(
    self,
    name: str | None = None,
    input: t.Any | None = None,
    output: t.Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: t.Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None:
    """
    Provide some useful default arguments when updating the current trace in our agents.
    """
    langfuse_context.update_current_trace(
        name=name,
        input=input,
        output=output,
        user_id=user_id or getpass.getuser(),
        session_id=session_id or self.session_id,
        # All traces within a single run execution will be grouped under a single session.
        version=version or APIKeys().LANGFUSE_DEPLOYMENT_VERSION,
        # Optionally, mark the current deployment with version (e.g. add git commit hash during docker building).
        release=release,
        metadata=metadata,
        tags=tags,
        public=public,
    )

load

load() -> None

Implement this method to load arbitrary instances needed across the whole run of the agent.

Do not customize init method.

Source code in prediction_market_agent_tooling/deploy/agent.py

def load(self) -> None:
    """
    Implement this method to load arbitrary instances needed across the whole run of the agent.

    Do not customize __init__ method.
    """

deploy_local

deploy_local(
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None

Run the agent in the forever cycle every sleep_time seconds, until the run_time is met.

Source code in prediction_market_agent_tooling/deploy/agent.py

def deploy_local(
    self,
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None:
    """
    Run the agent in the forever cycle every `sleep_time` seconds, until the `run_time` is met.
    """
    start_time = time.time()
    while run_time is None or time.time() - start_time < run_time:
        self.run(market_type=market_type)
        time.sleep(sleep_time)

get_gcloud_fname

get_gcloud_fname(market_type: MarketType) -> str

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_gcloud_fname(self, market_type: MarketType) -> str:
    return f"{self.__class__.__name__.lower()}-{market_type}-{utcnow().strftime('%Y-%m-%d--%H-%M-%S')}"

DeployableTraderAgent

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

Bases: DeployablePredictionAgent

Subclass this class to create your own prediction market trading agent.

The agent will process markets, make predictions and place trades (bets) based off these predictions.

Source code in prediction_market_agent_tooling/deploy/agent.py

def __init__(
    self,
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
    store_trades: bool = True,
    place_trades: bool = True,
) -> None:
    super().__init__(
        enable_langfuse=enable_langfuse, store_predictions=store_predictions
    )
    self.store_trades = store_trades
    self.place_trades = place_trades

AGENT_TAG class-attribute instance-attribute

AGENT_TAG: AgentTagEnum = TRADER

supported_markets class-attribute instance-attribute

supported_markets: Sequence[MarketType] = [
    OMEN,
    MANIFOLD,
    POLYMARKET,
    SEER,
]

store_trades instance-attribute

store_trades = store_trades

place_trades instance-attribute

place_trades = place_trades

start_time instance-attribute

start_time = utcnow()

enable_langfuse instance-attribute

enable_langfuse = enable_langfuse

api_keys instance-attribute

api_keys = APIKeys()

session_id cached property

session_id: str

bet_on_n_markets_per_run class-attribute instance-attribute

bet_on_n_markets_per_run: int = 1

n_markets_to_fetch class-attribute instance-attribute

n_markets_to_fetch: int = MAX_AVAILABLE_MARKETS

trade_on_markets_created_after class-attribute instance-attribute

trade_on_markets_created_after: DatetimeUTC | None = None

get_markets_sort_by class-attribute instance-attribute

get_markets_sort_by: SortBy = CLOSING_SOONEST

get_markets_filter_by class-attribute instance-attribute

get_markets_filter_by: FilterBy = OPEN

allow_invalid_questions class-attribute instance-attribute

allow_invalid_questions: bool = False

same_market_trade_interval class-attribute instance-attribute

same_market_trade_interval: TradeInterval = FixedInterval(
    timedelta(hours=24)
)

min_balance_to_keep_in_native_currency class-attribute instance-attribute

min_balance_to_keep_in_native_currency: xDai | None = (
    MINIMUM_NATIVE_TOKEN_IN_EOA_FOR_FEES
)

store_predictions instance-attribute

store_predictions = store_predictions

agent_name property

agent_name: str

fetch_categorical_markets property

fetch_categorical_markets: bool

initialize_langfuse

initialize_langfuse() -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def initialize_langfuse(self) -> None:
    super().initialize_langfuse()
    # Auto-observe all the methods where it makes sense, so that subclassses don't need to do it manually.
    self.build_trades = observe()(self.build_trades)  # type: ignore[method-assign]

check_min_required_balance_to_trade

check_min_required_balance_to_trade(
    market: AgentMarket,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_trade(self, market: AgentMarket) -> None:
    api_keys = APIKeys()

    # Get the strategy to know how much it will bet.
    strategy = self.get_betting_strategy(market)
    # Have a little bandwidth after the bet.
    min_required_balance_to_trade = strategy.maximum_possible_bet_amount * 1.01

    if market.get_trade_balance(api_keys) < min_required_balance_to_trade:
        raise OutOfFundsError(
            f"Minimum required balance {min_required_balance_to_trade} for agent {api_keys.bet_from_address=} is not met."
        )

get_total_amount_to_bet staticmethod

get_total_amount_to_bet(market: AgentMarket) -> USD

Source code in prediction_market_agent_tooling/deploy/agent.py

@staticmethod
def get_total_amount_to_bet(market: AgentMarket) -> USD:
    user_id = market.get_user_id(api_keys=APIKeys())

    total_amount = market.get_in_usd(market.get_tiny_bet_amount())
    existing_position = market.get_position(user_id=user_id)
    if existing_position and existing_position.total_amount_current > USD(0):
        total_amount += existing_position.total_amount_current
    return total_amount

get_betting_strategy

get_betting_strategy(
    market: AgentMarket,
) -> BettingStrategy

Override this method to customize betting strategy of your agent.

Given the market and prediction, agent uses this method to calculate optimal outcome and bet size.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_betting_strategy(self, market: AgentMarket) -> BettingStrategy:
    """
    Override this method to customize betting strategy of your agent.

    Given the market and prediction, agent uses this method to calculate optimal outcome and bet size.
    """
    total_amount = self.get_total_amount_to_bet(market)
    return MultiCategoricalMaxAccuracyBettingStrategy(bet_amount=total_amount)

build_trades

build_trades(
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
    existing_position: ExistingPosition | None,
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_trades(
    self,
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
    existing_position: ExistingPosition | None,
) -> list[Trade]:
    strategy = self.get_betting_strategy(market=market)
    trades = strategy.calculate_trades(existing_position, answer, market)
    return trades

before_process_market

before_process_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    super().before_process_market(market_type, market)
    self.check_min_required_balance_to_trade(market)

process_market

process_market(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedTradedMarket | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedTradedMarket | None:
    processed_market = super().process_market(market_type, market, verify_market)
    if processed_market is None:
        return None

    api_keys = APIKeys()
    user_id = market.get_user_id(api_keys=api_keys)

    existing_position = market.get_position(user_id=user_id)
    trades = self.build_trades(
        market=market,
        answer=processed_market.answer,
        existing_position=existing_position,
    )

    # It can take quite some time before agent processes all the markets, recheck here if the market didn't get closed in the meantime, to not error out completely.
    # Unfortunately, we can not just add some room into closing time of the market while fetching them, because liquidity can be removed at any time by the liquidity providers.
    still_tradeable = market.can_be_traded()
    if not still_tradeable:
        logger.warning(
            f"Market {market.question=} ({market.url}) was selected to processing, but is not tradeable anymore."
        )

    placed_trades = []
    for trade in trades:
        logger.info(f"Executing trade {trade} on market {market.id} ({market.url})")

        if self.place_trades and still_tradeable:
            match trade.trade_type:
                case TradeType.BUY:
                    id = market.buy_tokens(
                        outcome=trade.outcome, amount=trade.amount
                    )
                case TradeType.SELL:
                    # Get actual value of the position we are going to sell, and if it's less than we wanted to sell, simply sell all of it.
                    current_position = check_not_none(
                        market.get_position(user_id),
                        "Should exists if we are going to sell outcomes.",
                    )

                    current_position_value_usd = current_position.amounts_current[
                        trade.outcome
                    ]
                    amount_to_sell: USD | OutcomeToken
                    if current_position_value_usd <= trade.amount:
                        logger.warning(
                            f"Current value of position {trade.outcome=}, {current_position_value_usd=} is less than the desired selling amount {trade.amount=}. Selling all."
                        )
                        # In case the agent asked to sell too much, provide the amount to sell as all outcome tokens, instead of in USD, to minimze fx fluctuations when selling.
                        amount_to_sell = current_position.amounts_ot[trade.outcome]
                    else:
                        amount_to_sell = trade.amount
                    id = market.sell_tokens(
                        outcome=trade.outcome,
                        amount=amount_to_sell,
                    )
                case _:
                    raise ValueError(f"Unexpected trade type {trade.trade_type}.")
            placed_trades.append(PlacedTrade.from_trade(trade, id))
        else:
            logger.info(
                f"Trade execution skipped because, {self.place_trades=} or {still_tradeable=}."
            )

    traded_market = ProcessedTradedMarket(
        answer=processed_market.answer, trades=placed_trades
    )
    logger.info(f"Traded market {market.question=} from {market.url=}.")
    return traded_market

after_process_market

after_process_market(
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None:
    api_keys = APIKeys()
    super().after_process_market(
        market_type,
        market,
        processed_market,
    )
    if isinstance(processed_market, ProcessedTradedMarket):
        if self.store_trades:
            market.store_trades(processed_market, api_keys, self.agent_name)
        else:
            logger.info(
                f"Trades {processed_market.trades} not stored because {self.store_trades=}."
            )

langfuse_update_current_trace

langfuse_update_current_trace(
    name: str | None = None,
    input: Any | None = None,
    output: Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None

Provide some useful default arguments when updating the current trace in our agents.

Source code in prediction_market_agent_tooling/deploy/agent.py

def langfuse_update_current_trace(
    self,
    name: str | None = None,
    input: t.Any | None = None,
    output: t.Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: t.Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None:
    """
    Provide some useful default arguments when updating the current trace in our agents.
    """
    langfuse_context.update_current_trace(
        name=name,
        input=input,
        output=output,
        user_id=user_id or getpass.getuser(),
        session_id=session_id or self.session_id,
        # All traces within a single run execution will be grouped under a single session.
        version=version or APIKeys().LANGFUSE_DEPLOYMENT_VERSION,
        # Optionally, mark the current deployment with version (e.g. add git commit hash during docker building).
        release=release,
        metadata=metadata,
        tags=tags,
        public=public,
    )

load

load() -> None

Implement this method to load arbitrary instances needed across the whole run of the agent.

Do not customize init method.

Source code in prediction_market_agent_tooling/deploy/agent.py

def load(self) -> None:
    """
    Implement this method to load arbitrary instances needed across the whole run of the agent.

    Do not customize __init__ method.
    """

deploy_local

deploy_local(
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None

Run the agent in the forever cycle every sleep_time seconds, until the run_time is met.

Source code in prediction_market_agent_tooling/deploy/agent.py

def deploy_local(
    self,
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None:
    """
    Run the agent in the forever cycle every `sleep_time` seconds, until the `run_time` is met.
    """
    start_time = time.time()
    while run_time is None or time.time() - start_time < run_time:
        self.run(market_type=market_type)
        time.sleep(sleep_time)

run

run(market_type: MarketType) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def run(self, market_type: MarketType) -> None:
    if market_type not in self.supported_markets:
        raise ValueError(
            f"Only {self.supported_markets} are supported by this agent."
        )
    self.before_process_markets(market_type)
    self.process_markets(market_type)
    self.after_process_markets(market_type)

get_gcloud_fname

get_gcloud_fname(market_type: MarketType) -> str

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_gcloud_fname(self, market_type: MarketType) -> str:
    return f"{self.__class__.__name__.lower()}-{market_type}-{utcnow().strftime('%Y-%m-%d--%H-%M-%S')}"

update_langfuse_trace_by_market

update_langfuse_trace_by_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    self.langfuse_update_current_trace(
        # UI allows to do filtering by these.
        metadata={
            "agent_class": self.__class__.__name__,
            "market_id": market.id,
            "market_question": market.question,
            "market_outcomes": market.outcomes,
        },
    )

update_langfuse_trace_by_processed_market

update_langfuse_trace_by_processed_market(
    market_type: MarketType,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_processed_market(
    self, market_type: MarketType, processed_market: ProcessedMarket | None
) -> None:
    self.langfuse_update_current_trace(
        tags=[
            self.AGENT_TAG,
            (
                AnsweredEnum.ANSWERED
                if processed_market is not None
                else AnsweredEnum.NOT_ANSWERED
            ),
            market_type.value,
        ]
    )

check_min_required_balance_to_operate

check_min_required_balance_to_operate(
    market_type: MarketType,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_operate(self, market_type: MarketType) -> None:
    api_keys = APIKeys()

    if not market_type.market_class.verify_operational_balance(api_keys):
        raise CantPayForGasError(
            f"{api_keys=} doesn't have enough operational balance."
        )

verify_market

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

Subclasses can implement their own logic instead of this one, or on top of this one. By default, it allows only markets where user didn't bet recently and it's a reasonable question.

Source code in prediction_market_agent_tooling/deploy/agent.py

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    """
    Subclasses can implement their own logic instead of this one, or on top of this one.
    By default, it allows only markets where user didn't bet recently and it's a reasonable question.
    """
    if market.have_bet_on_market_since(
        keys=APIKeys(), since=self.same_market_trade_interval.get(market=market)
    ):
        logger.info(
            f"Market already bet on within {self.same_market_trade_interval}."
        )
        return False

    # Manifold allows to bet only on markets with probability between 1 and 99.
    if market_type == MarketType.MANIFOLD:
        probability_yes = market.probabilities[
            market.get_outcome_str_from_bool(True)
        ]
        if not probability_yes or not 1 < probability_yes < 99:
            logger.info("Manifold's market probability not in the range 1-99.")
            return False

    # Do as a last check, as it uses paid OpenAI API.
    if not is_predictable_binary(market.question):
        logger.info("Market question is not predictable.")
        return False

    if not self.allow_invalid_questions and is_invalid(market.question):
        logger.info("Market question is invalid.")
        return False

    return True

answer_categorical_market

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

Source code in prediction_market_agent_tooling/deploy/agent.py

def answer_categorical_market(
    self, market: AgentMarket
) -> CategoricalProbabilisticAnswer | None:
    raise NotImplementedError("This method must be implemented by the subclass")

answer_binary_market

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

Answer the binary market.

If this method is not overridden by the subclass, it will fall back to using answer_categorical_market(). Therefore, subclasses only need to implement answer_categorical_market() if they want to handle both types of markets.

Source code in prediction_market_agent_tooling/deploy/agent.py

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    """
    Answer the binary market.

    If this method is not overridden by the subclass, it will fall back to using
    answer_categorical_market(). Therefore, subclasses only need to implement
    answer_categorical_market() if they want to handle both types of markets.
    """
    raise NotImplementedError(
        "Either this method, or answer_categorical_market, must be implemented by the subclass."
    )

get_markets

get_markets(
    market_type: MarketType,
) -> t.Sequence[AgentMarket]

Override this method to customize what markets will fetch for processing.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_markets(
    self,
    market_type: MarketType,
) -> t.Sequence[AgentMarket]:
    """
    Override this method to customize what markets will fetch for processing.
    """
    cls = market_type.market_class
    # Fetch the soonest closing markets to choose from
    available_markets = cls.get_markets(
        limit=self.n_markets_to_fetch,
        sort_by=self.get_markets_sort_by,
        filter_by=self.get_markets_filter_by,
        created_after=self.trade_on_markets_created_after,
        fetch_categorical_markets=self.fetch_categorical_markets,
    )
    return available_markets

build_answer

build_answer(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_answer(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None:
    if verify_market and not self.verify_market(market_type, market):
        logger.info(f"Market '{market.question}' doesn't meet the criteria.")
        return None

    logger.info(f"Answering market '{market.question}'.")

    if market.is_binary:
        try:
            binary_answer = self.answer_binary_market(market)
            return (
                CategoricalProbabilisticAnswer.from_probabilistic_answer(
                    binary_answer
                )
                if binary_answer is not None
                else None
            )
        except NotImplementedError:
            logger.info(
                "answer_binary_market() not implemented, falling back to answer_categorical_market()"
            )

    return self.answer_categorical_market(market)

before_process_markets

before_process_markets(market_type: MarketType) -> None

Executed before market processing loop starts.

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_markets(self, market_type: MarketType) -> None:
    """
    Executed before market processing loop starts.
    """
    api_keys = APIKeys()
    self.check_min_required_balance_to_operate(market_type)
    market_type.market_class.redeem_winnings(api_keys)

process_markets

process_markets(market_type: MarketType) -> None

Processes bets placed by agents on a given market.

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_markets(self, market_type: MarketType) -> None:
    """
    Processes bets placed by agents on a given market.
    """
    logger.info("Start processing of markets.")
    available_markets = self.get_markets(market_type)

    logger.info(
        f"Fetched {len(available_markets)=} markets to process, going to process {self.bet_on_n_markets_per_run=}."
    )
    processed = 0

    for market_idx, market in enumerate(available_markets):
        logger.info(
            f"Going to process market {market.url}: {market_idx+1} / {len(available_markets)}."
        )
        self.before_process_market(market_type, market)
        processed_market = self.process_market(market_type, market)
        self.after_process_market(market_type, market, processed_market)

        if processed_market is not None:
            processed += 1

        if processed == self.bet_on_n_markets_per_run:
            break

    logger.info(
        f"All markets processed. Successfully processed {processed}/{len(available_markets)}."
    )

after_process_markets

after_process_markets(market_type: MarketType) -> None

Executed after market processing loop ends.

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_markets(self, market_type: MarketType) -> None:
    """
    Executed after market processing loop ends.
    """

initialize_langfuse

initialize_langfuse(enable_langfuse: bool) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def initialize_langfuse(enable_langfuse: bool) -> None:
    # Configure Langfuse singleton with our APIKeys.
    # If langfuse is disabled, it will just ignore all the calls, so no need to do if-else around the code.
    keys = APIKeys()
    if enable_langfuse:
        langfuse_context.configure(
            public_key=keys.langfuse_public_key,
            secret_key=keys.langfuse_secret_key.get_secret_value(),
            host=keys.langfuse_host,
            enabled=enable_langfuse,
        )
    else:
        langfuse_context.configure(enabled=enable_langfuse)

agent_example

DeployableCoinFlipAgent

DeployableCoinFlipAgent(
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
    store_trades: bool = True,
    place_trades: bool = True,
)

Bases: DeployableTraderAgent

Source code in prediction_market_agent_tooling/deploy/agent.py

def __init__(
    self,
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
    store_trades: bool = True,
    place_trades: bool = True,
) -> None:
    super().__init__(
        enable_langfuse=enable_langfuse, store_predictions=store_predictions
    )
    self.store_trades = store_trades
    self.place_trades = place_trades

get_markets_sort_by class-attribute instance-attribute

get_markets_sort_by = HIGHEST_LIQUIDITY

start_time instance-attribute

start_time = utcnow()

enable_langfuse instance-attribute

enable_langfuse = enable_langfuse

api_keys instance-attribute

api_keys = APIKeys()

session_id cached property

session_id: str

AGENT_TAG class-attribute instance-attribute

AGENT_TAG: AgentTagEnum = TRADER

bet_on_n_markets_per_run class-attribute instance-attribute

bet_on_n_markets_per_run: int = 1

n_markets_to_fetch class-attribute instance-attribute

n_markets_to_fetch: int = MAX_AVAILABLE_MARKETS

trade_on_markets_created_after class-attribute instance-attribute

trade_on_markets_created_after: DatetimeUTC | None = None

get_markets_filter_by class-attribute instance-attribute

get_markets_filter_by: FilterBy = OPEN

allow_invalid_questions class-attribute instance-attribute

allow_invalid_questions: bool = False

same_market_trade_interval class-attribute instance-attribute

same_market_trade_interval: TradeInterval = FixedInterval(
    timedelta(hours=24)
)

min_balance_to_keep_in_native_currency class-attribute instance-attribute

min_balance_to_keep_in_native_currency: xDai | None = (
    MINIMUM_NATIVE_TOKEN_IN_EOA_FOR_FEES
)

supported_markets class-attribute instance-attribute

supported_markets: Sequence[MarketType] = [
    OMEN,
    MANIFOLD,
    POLYMARKET,
    SEER,
]

store_predictions instance-attribute

store_predictions = store_predictions

agent_name property

agent_name: str

fetch_categorical_markets property

fetch_categorical_markets: bool

store_trades instance-attribute

store_trades = store_trades

place_trades instance-attribute

place_trades = place_trades

verify_market

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

Source code in prediction_market_agent_tooling/deploy/agent_example.py

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

answer_categorical_market

answer_categorical_market(
    market: AgentMarket,
) -> CategoricalProbabilisticAnswer

Source code in prediction_market_agent_tooling/deploy/agent_example.py

def answer_categorical_market(
    self, market: AgentMarket
) -> CategoricalProbabilisticAnswer:
    decision = random.choice(market.outcomes)
    probabilities = {decision: Probability(1.0)}
    for outcome in market.outcomes:
        if outcome != decision:
            probabilities[outcome] = Probability(0.0)
    return CategoricalProbabilisticAnswer(
        probabilities=probabilities,
        confidence=0.5,
        reasoning="I flipped a coin to decide.",
    )

initialize_langfuse

initialize_langfuse() -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def initialize_langfuse(self) -> None:
    super().initialize_langfuse()
    # Auto-observe all the methods where it makes sense, so that subclassses don't need to do it manually.
    self.build_trades = observe()(self.build_trades)  # type: ignore[method-assign]

langfuse_update_current_trace

langfuse_update_current_trace(
    name: str | None = None,
    input: Any | None = None,
    output: Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None

Provide some useful default arguments when updating the current trace in our agents.

Source code in prediction_market_agent_tooling/deploy/agent.py

def langfuse_update_current_trace(
    self,
    name: str | None = None,
    input: t.Any | None = None,
    output: t.Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: t.Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None:
    """
    Provide some useful default arguments when updating the current trace in our agents.
    """
    langfuse_context.update_current_trace(
        name=name,
        input=input,
        output=output,
        user_id=user_id or getpass.getuser(),
        session_id=session_id or self.session_id,
        # All traces within a single run execution will be grouped under a single session.
        version=version or APIKeys().LANGFUSE_DEPLOYMENT_VERSION,
        # Optionally, mark the current deployment with version (e.g. add git commit hash during docker building).
        release=release,
        metadata=metadata,
        tags=tags,
        public=public,
    )

load

load() -> None

Implement this method to load arbitrary instances needed across the whole run of the agent.

Do not customize init method.

Source code in prediction_market_agent_tooling/deploy/agent.py

def load(self) -> None:
    """
    Implement this method to load arbitrary instances needed across the whole run of the agent.

    Do not customize __init__ method.
    """

deploy_local

deploy_local(
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None

Run the agent in the forever cycle every sleep_time seconds, until the run_time is met.

Source code in prediction_market_agent_tooling/deploy/agent.py

def deploy_local(
    self,
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None:
    """
    Run the agent in the forever cycle every `sleep_time` seconds, until the `run_time` is met.
    """
    start_time = time.time()
    while run_time is None or time.time() - start_time < run_time:
        self.run(market_type=market_type)
        time.sleep(sleep_time)

run

run(market_type: MarketType) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def run(self, market_type: MarketType) -> None:
    if market_type not in self.supported_markets:
        raise ValueError(
            f"Only {self.supported_markets} are supported by this agent."
        )
    self.before_process_markets(market_type)
    self.process_markets(market_type)
    self.after_process_markets(market_type)

get_gcloud_fname

get_gcloud_fname(market_type: MarketType) -> str

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_gcloud_fname(self, market_type: MarketType) -> str:
    return f"{self.__class__.__name__.lower()}-{market_type}-{utcnow().strftime('%Y-%m-%d--%H-%M-%S')}"

update_langfuse_trace_by_market

update_langfuse_trace_by_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    self.langfuse_update_current_trace(
        # UI allows to do filtering by these.
        metadata={
            "agent_class": self.__class__.__name__,
            "market_id": market.id,
            "market_question": market.question,
            "market_outcomes": market.outcomes,
        },
    )

update_langfuse_trace_by_processed_market

update_langfuse_trace_by_processed_market(
    market_type: MarketType,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_processed_market(
    self, market_type: MarketType, processed_market: ProcessedMarket | None
) -> None:
    self.langfuse_update_current_trace(
        tags=[
            self.AGENT_TAG,
            (
                AnsweredEnum.ANSWERED
                if processed_market is not None
                else AnsweredEnum.NOT_ANSWERED
            ),
            market_type.value,
        ]
    )

check_min_required_balance_to_operate

check_min_required_balance_to_operate(
    market_type: MarketType,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_operate(self, market_type: MarketType) -> None:
    api_keys = APIKeys()

    if not market_type.market_class.verify_operational_balance(api_keys):
        raise CantPayForGasError(
            f"{api_keys=} doesn't have enough operational balance."
        )

answer_binary_market

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

Answer the binary market.

If this method is not overridden by the subclass, it will fall back to using answer_categorical_market(). Therefore, subclasses only need to implement answer_categorical_market() if they want to handle both types of markets.

Source code in prediction_market_agent_tooling/deploy/agent.py

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    """
    Answer the binary market.

    If this method is not overridden by the subclass, it will fall back to using
    answer_categorical_market(). Therefore, subclasses only need to implement
    answer_categorical_market() if they want to handle both types of markets.
    """
    raise NotImplementedError(
        "Either this method, or answer_categorical_market, must be implemented by the subclass."
    )

get_markets

get_markets(
    market_type: MarketType,
) -> t.Sequence[AgentMarket]

Override this method to customize what markets will fetch for processing.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_markets(
    self,
    market_type: MarketType,
) -> t.Sequence[AgentMarket]:
    """
    Override this method to customize what markets will fetch for processing.
    """
    cls = market_type.market_class
    # Fetch the soonest closing markets to choose from
    available_markets = cls.get_markets(
        limit=self.n_markets_to_fetch,
        sort_by=self.get_markets_sort_by,
        filter_by=self.get_markets_filter_by,
        created_after=self.trade_on_markets_created_after,
        fetch_categorical_markets=self.fetch_categorical_markets,
    )
    return available_markets

before_process_market

before_process_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    super().before_process_market(market_type, market)
    self.check_min_required_balance_to_trade(market)

build_answer

build_answer(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_answer(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None:
    if verify_market and not self.verify_market(market_type, market):
        logger.info(f"Market '{market.question}' doesn't meet the criteria.")
        return None

    logger.info(f"Answering market '{market.question}'.")

    if market.is_binary:
        try:
            binary_answer = self.answer_binary_market(market)
            return (
                CategoricalProbabilisticAnswer.from_probabilistic_answer(
                    binary_answer
                )
                if binary_answer is not None
                else None
            )
        except NotImplementedError:
            logger.info(
                "answer_binary_market() not implemented, falling back to answer_categorical_market()"
            )

    return self.answer_categorical_market(market)

process_market

process_market(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedTradedMarket | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedTradedMarket | None:
    processed_market = super().process_market(market_type, market, verify_market)
    if processed_market is None:
        return None

    api_keys = APIKeys()
    user_id = market.get_user_id(api_keys=api_keys)

    existing_position = market.get_position(user_id=user_id)
    trades = self.build_trades(
        market=market,
        answer=processed_market.answer,
        existing_position=existing_position,
    )

    # It can take quite some time before agent processes all the markets, recheck here if the market didn't get closed in the meantime, to not error out completely.
    # Unfortunately, we can not just add some room into closing time of the market while fetching them, because liquidity can be removed at any time by the liquidity providers.
    still_tradeable = market.can_be_traded()
    if not still_tradeable:
        logger.warning(
            f"Market {market.question=} ({market.url}) was selected to processing, but is not tradeable anymore."
        )

    placed_trades = []
    for trade in trades:
        logger.info(f"Executing trade {trade} on market {market.id} ({market.url})")

        if self.place_trades and still_tradeable:
            match trade.trade_type:
                case TradeType.BUY:
                    id = market.buy_tokens(
                        outcome=trade.outcome, amount=trade.amount
                    )
                case TradeType.SELL:
                    # Get actual value of the position we are going to sell, and if it's less than we wanted to sell, simply sell all of it.
                    current_position = check_not_none(
                        market.get_position(user_id),
                        "Should exists if we are going to sell outcomes.",
                    )

                    current_position_value_usd = current_position.amounts_current[
                        trade.outcome
                    ]
                    amount_to_sell: USD | OutcomeToken
                    if current_position_value_usd <= trade.amount:
                        logger.warning(
                            f"Current value of position {trade.outcome=}, {current_position_value_usd=} is less than the desired selling amount {trade.amount=}. Selling all."
                        )
                        # In case the agent asked to sell too much, provide the amount to sell as all outcome tokens, instead of in USD, to minimze fx fluctuations when selling.
                        amount_to_sell = current_position.amounts_ot[trade.outcome]
                    else:
                        amount_to_sell = trade.amount
                    id = market.sell_tokens(
                        outcome=trade.outcome,
                        amount=amount_to_sell,
                    )
                case _:
                    raise ValueError(f"Unexpected trade type {trade.trade_type}.")
            placed_trades.append(PlacedTrade.from_trade(trade, id))
        else:
            logger.info(
                f"Trade execution skipped because, {self.place_trades=} or {still_tradeable=}."
            )

    traded_market = ProcessedTradedMarket(
        answer=processed_market.answer, trades=placed_trades
    )
    logger.info(f"Traded market {market.question=} from {market.url=}.")
    return traded_market

after_process_market

after_process_market(
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None:
    api_keys = APIKeys()
    super().after_process_market(
        market_type,
        market,
        processed_market,
    )
    if isinstance(processed_market, ProcessedTradedMarket):
        if self.store_trades:
            market.store_trades(processed_market, api_keys, self.agent_name)
        else:
            logger.info(
                f"Trades {processed_market.trades} not stored because {self.store_trades=}."
            )

before_process_markets

before_process_markets(market_type: MarketType) -> None

Executed before market processing loop starts.

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_markets(self, market_type: MarketType) -> None:
    """
    Executed before market processing loop starts.
    """
    api_keys = APIKeys()
    self.check_min_required_balance_to_operate(market_type)
    market_type.market_class.redeem_winnings(api_keys)

process_markets

process_markets(market_type: MarketType) -> None

Processes bets placed by agents on a given market.

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_markets(self, market_type: MarketType) -> None:
    """
    Processes bets placed by agents on a given market.
    """
    logger.info("Start processing of markets.")
    available_markets = self.get_markets(market_type)

    logger.info(
        f"Fetched {len(available_markets)=} markets to process, going to process {self.bet_on_n_markets_per_run=}."
    )
    processed = 0

    for market_idx, market in enumerate(available_markets):
        logger.info(
            f"Going to process market {market.url}: {market_idx+1} / {len(available_markets)}."
        )
        self.before_process_market(market_type, market)
        processed_market = self.process_market(market_type, market)
        self.after_process_market(market_type, market, processed_market)

        if processed_market is not None:
            processed += 1

        if processed == self.bet_on_n_markets_per_run:
            break

    logger.info(
        f"All markets processed. Successfully processed {processed}/{len(available_markets)}."
    )

after_process_markets

after_process_markets(market_type: MarketType) -> None

Executed after market processing loop ends.

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_markets(self, market_type: MarketType) -> None:
    """
    Executed after market processing loop ends.
    """

check_min_required_balance_to_trade

check_min_required_balance_to_trade(
    market: AgentMarket,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_trade(self, market: AgentMarket) -> None:
    api_keys = APIKeys()

    # Get the strategy to know how much it will bet.
    strategy = self.get_betting_strategy(market)
    # Have a little bandwidth after the bet.
    min_required_balance_to_trade = strategy.maximum_possible_bet_amount * 1.01

    if market.get_trade_balance(api_keys) < min_required_balance_to_trade:
        raise OutOfFundsError(
            f"Minimum required balance {min_required_balance_to_trade} for agent {api_keys.bet_from_address=} is not met."
        )

get_total_amount_to_bet staticmethod

get_total_amount_to_bet(market: AgentMarket) -> USD

Source code in prediction_market_agent_tooling/deploy/agent.py

@staticmethod
def get_total_amount_to_bet(market: AgentMarket) -> USD:
    user_id = market.get_user_id(api_keys=APIKeys())

    total_amount = market.get_in_usd(market.get_tiny_bet_amount())
    existing_position = market.get_position(user_id=user_id)
    if existing_position and existing_position.total_amount_current > USD(0):
        total_amount += existing_position.total_amount_current
    return total_amount

get_betting_strategy

get_betting_strategy(
    market: AgentMarket,
) -> BettingStrategy

Override this method to customize betting strategy of your agent.

Given the market and prediction, agent uses this method to calculate optimal outcome and bet size.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_betting_strategy(self, market: AgentMarket) -> BettingStrategy:
    """
    Override this method to customize betting strategy of your agent.

    Given the market and prediction, agent uses this method to calculate optimal outcome and bet size.
    """
    total_amount = self.get_total_amount_to_bet(market)
    return MultiCategoricalMaxAccuracyBettingStrategy(bet_amount=total_amount)

build_trades

build_trades(
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
    existing_position: ExistingPosition | None,
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_trades(
    self,
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
    existing_position: ExistingPosition | None,
) -> list[Trade]:
    strategy = self.get_betting_strategy(market=market)
    trades = strategy.calculate_trades(existing_position, answer, market)
    return trades

DeployableAlwaysRaiseAgent

DeployableAlwaysRaiseAgent(
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
    store_trades: bool = True,
    place_trades: bool = True,
)

Bases: DeployableTraderAgent

Source code in prediction_market_agent_tooling/deploy/agent.py

def __init__(
    self,
    enable_langfuse: bool = APIKeys().default_enable_langfuse,
    store_predictions: bool = True,
    store_trades: bool = True,
    place_trades: bool = True,
) -> None:
    super().__init__(
        enable_langfuse=enable_langfuse, store_predictions=store_predictions
    )
    self.store_trades = store_trades
    self.place_trades = place_trades

start_time instance-attribute

start_time = utcnow()

enable_langfuse instance-attribute

enable_langfuse = enable_langfuse

api_keys instance-attribute

api_keys = APIKeys()

session_id cached property

session_id: str

AGENT_TAG class-attribute instance-attribute

AGENT_TAG: AgentTagEnum = TRADER

bet_on_n_markets_per_run class-attribute instance-attribute

bet_on_n_markets_per_run: int = 1

n_markets_to_fetch class-attribute instance-attribute

n_markets_to_fetch: int = MAX_AVAILABLE_MARKETS

trade_on_markets_created_after class-attribute instance-attribute

trade_on_markets_created_after: DatetimeUTC | None = None

get_markets_sort_by class-attribute instance-attribute

get_markets_sort_by: SortBy = CLOSING_SOONEST

get_markets_filter_by class-attribute instance-attribute

get_markets_filter_by: FilterBy = OPEN

allow_invalid_questions class-attribute instance-attribute

allow_invalid_questions: bool = False

same_market_trade_interval class-attribute instance-attribute

same_market_trade_interval: TradeInterval = FixedInterval(
    timedelta(hours=24)
)

min_balance_to_keep_in_native_currency class-attribute instance-attribute

min_balance_to_keep_in_native_currency: xDai | None = (
    MINIMUM_NATIVE_TOKEN_IN_EOA_FOR_FEES
)

supported_markets class-attribute instance-attribute

supported_markets: Sequence[MarketType] = [
    OMEN,
    MANIFOLD,
    POLYMARKET,
    SEER,
]

store_predictions instance-attribute

store_predictions = store_predictions

agent_name property

agent_name: str

fetch_categorical_markets property

fetch_categorical_markets: bool

store_trades instance-attribute

store_trades = store_trades

place_trades instance-attribute

place_trades = place_trades

answer_categorical_market

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

Source code in prediction_market_agent_tooling/deploy/agent_example.py

def answer_categorical_market(
    self, market: AgentMarket
) -> CategoricalProbabilisticAnswer | None:
    raise RuntimeError("I always raise!")

initialize_langfuse

initialize_langfuse() -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def initialize_langfuse(self) -> None:
    super().initialize_langfuse()
    # Auto-observe all the methods where it makes sense, so that subclassses don't need to do it manually.
    self.build_trades = observe()(self.build_trades)  # type: ignore[method-assign]

langfuse_update_current_trace

langfuse_update_current_trace(
    name: str | None = None,
    input: Any | None = None,
    output: Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None

Provide some useful default arguments when updating the current trace in our agents.

Source code in prediction_market_agent_tooling/deploy/agent.py

def langfuse_update_current_trace(
    self,
    name: str | None = None,
    input: t.Any | None = None,
    output: t.Any | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
    version: str | None = None,
    release: str | None = None,
    metadata: t.Any | None = None,
    tags: list[str] | None = None,
    public: bool | None = None,
) -> None:
    """
    Provide some useful default arguments when updating the current trace in our agents.
    """
    langfuse_context.update_current_trace(
        name=name,
        input=input,
        output=output,
        user_id=user_id or getpass.getuser(),
        session_id=session_id or self.session_id,
        # All traces within a single run execution will be grouped under a single session.
        version=version or APIKeys().LANGFUSE_DEPLOYMENT_VERSION,
        # Optionally, mark the current deployment with version (e.g. add git commit hash during docker building).
        release=release,
        metadata=metadata,
        tags=tags,
        public=public,
    )

load

load() -> None

Implement this method to load arbitrary instances needed across the whole run of the agent.

Do not customize init method.

Source code in prediction_market_agent_tooling/deploy/agent.py

def load(self) -> None:
    """
    Implement this method to load arbitrary instances needed across the whole run of the agent.

    Do not customize __init__ method.
    """

deploy_local

deploy_local(
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None

Run the agent in the forever cycle every sleep_time seconds, until the run_time is met.

Source code in prediction_market_agent_tooling/deploy/agent.py

def deploy_local(
    self,
    market_type: MarketType,
    sleep_time: float,
    run_time: float | None,
) -> None:
    """
    Run the agent in the forever cycle every `sleep_time` seconds, until the `run_time` is met.
    """
    start_time = time.time()
    while run_time is None or time.time() - start_time < run_time:
        self.run(market_type=market_type)
        time.sleep(sleep_time)

run

run(market_type: MarketType) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def run(self, market_type: MarketType) -> None:
    if market_type not in self.supported_markets:
        raise ValueError(
            f"Only {self.supported_markets} are supported by this agent."
        )
    self.before_process_markets(market_type)
    self.process_markets(market_type)
    self.after_process_markets(market_type)

get_gcloud_fname

get_gcloud_fname(market_type: MarketType) -> str

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_gcloud_fname(self, market_type: MarketType) -> str:
    return f"{self.__class__.__name__.lower()}-{market_type}-{utcnow().strftime('%Y-%m-%d--%H-%M-%S')}"

update_langfuse_trace_by_market

update_langfuse_trace_by_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    self.langfuse_update_current_trace(
        # UI allows to do filtering by these.
        metadata={
            "agent_class": self.__class__.__name__,
            "market_id": market.id,
            "market_question": market.question,
            "market_outcomes": market.outcomes,
        },
    )

update_langfuse_trace_by_processed_market

update_langfuse_trace_by_processed_market(
    market_type: MarketType,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def update_langfuse_trace_by_processed_market(
    self, market_type: MarketType, processed_market: ProcessedMarket | None
) -> None:
    self.langfuse_update_current_trace(
        tags=[
            self.AGENT_TAG,
            (
                AnsweredEnum.ANSWERED
                if processed_market is not None
                else AnsweredEnum.NOT_ANSWERED
            ),
            market_type.value,
        ]
    )

check_min_required_balance_to_operate

check_min_required_balance_to_operate(
    market_type: MarketType,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_operate(self, market_type: MarketType) -> None:
    api_keys = APIKeys()

    if not market_type.market_class.verify_operational_balance(api_keys):
        raise CantPayForGasError(
            f"{api_keys=} doesn't have enough operational balance."
        )

verify_market

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

Subclasses can implement their own logic instead of this one, or on top of this one. By default, it allows only markets where user didn't bet recently and it's a reasonable question.

Source code in prediction_market_agent_tooling/deploy/agent.py

def verify_market(self, market_type: MarketType, market: AgentMarket) -> bool:
    """
    Subclasses can implement their own logic instead of this one, or on top of this one.
    By default, it allows only markets where user didn't bet recently and it's a reasonable question.
    """
    if market.have_bet_on_market_since(
        keys=APIKeys(), since=self.same_market_trade_interval.get(market=market)
    ):
        logger.info(
            f"Market already bet on within {self.same_market_trade_interval}."
        )
        return False

    # Manifold allows to bet only on markets with probability between 1 and 99.
    if market_type == MarketType.MANIFOLD:
        probability_yes = market.probabilities[
            market.get_outcome_str_from_bool(True)
        ]
        if not probability_yes or not 1 < probability_yes < 99:
            logger.info("Manifold's market probability not in the range 1-99.")
            return False

    # Do as a last check, as it uses paid OpenAI API.
    if not is_predictable_binary(market.question):
        logger.info("Market question is not predictable.")
        return False

    if not self.allow_invalid_questions and is_invalid(market.question):
        logger.info("Market question is invalid.")
        return False

    return True

answer_binary_market

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

Answer the binary market.

If this method is not overridden by the subclass, it will fall back to using answer_categorical_market(). Therefore, subclasses only need to implement answer_categorical_market() if they want to handle both types of markets.

Source code in prediction_market_agent_tooling/deploy/agent.py

def answer_binary_market(self, market: AgentMarket) -> ProbabilisticAnswer | None:
    """
    Answer the binary market.

    If this method is not overridden by the subclass, it will fall back to using
    answer_categorical_market(). Therefore, subclasses only need to implement
    answer_categorical_market() if they want to handle both types of markets.
    """
    raise NotImplementedError(
        "Either this method, or answer_categorical_market, must be implemented by the subclass."
    )

get_markets

get_markets(
    market_type: MarketType,
) -> t.Sequence[AgentMarket]

Override this method to customize what markets will fetch for processing.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_markets(
    self,
    market_type: MarketType,
) -> t.Sequence[AgentMarket]:
    """
    Override this method to customize what markets will fetch for processing.
    """
    cls = market_type.market_class
    # Fetch the soonest closing markets to choose from
    available_markets = cls.get_markets(
        limit=self.n_markets_to_fetch,
        sort_by=self.get_markets_sort_by,
        filter_by=self.get_markets_filter_by,
        created_after=self.trade_on_markets_created_after,
        fetch_categorical_markets=self.fetch_categorical_markets,
    )
    return available_markets

before_process_market

before_process_market(
    market_type: MarketType, market: AgentMarket
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_market(
    self, market_type: MarketType, market: AgentMarket
) -> None:
    super().before_process_market(market_type, market)
    self.check_min_required_balance_to_trade(market)

build_answer

build_answer(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_answer(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> CategoricalProbabilisticAnswer | None:
    if verify_market and not self.verify_market(market_type, market):
        logger.info(f"Market '{market.question}' doesn't meet the criteria.")
        return None

    logger.info(f"Answering market '{market.question}'.")

    if market.is_binary:
        try:
            binary_answer = self.answer_binary_market(market)
            return (
                CategoricalProbabilisticAnswer.from_probabilistic_answer(
                    binary_answer
                )
                if binary_answer is not None
                else None
            )
        except NotImplementedError:
            logger.info(
                "answer_binary_market() not implemented, falling back to answer_categorical_market()"
            )

    return self.answer_categorical_market(market)

process_market

process_market(
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedTradedMarket | None

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    verify_market: bool = True,
) -> ProcessedTradedMarket | None:
    processed_market = super().process_market(market_type, market, verify_market)
    if processed_market is None:
        return None

    api_keys = APIKeys()
    user_id = market.get_user_id(api_keys=api_keys)

    existing_position = market.get_position(user_id=user_id)
    trades = self.build_trades(
        market=market,
        answer=processed_market.answer,
        existing_position=existing_position,
    )

    # It can take quite some time before agent processes all the markets, recheck here if the market didn't get closed in the meantime, to not error out completely.
    # Unfortunately, we can not just add some room into closing time of the market while fetching them, because liquidity can be removed at any time by the liquidity providers.
    still_tradeable = market.can_be_traded()
    if not still_tradeable:
        logger.warning(
            f"Market {market.question=} ({market.url}) was selected to processing, but is not tradeable anymore."
        )

    placed_trades = []
    for trade in trades:
        logger.info(f"Executing trade {trade} on market {market.id} ({market.url})")

        if self.place_trades and still_tradeable:
            match trade.trade_type:
                case TradeType.BUY:
                    id = market.buy_tokens(
                        outcome=trade.outcome, amount=trade.amount
                    )
                case TradeType.SELL:
                    # Get actual value of the position we are going to sell, and if it's less than we wanted to sell, simply sell all of it.
                    current_position = check_not_none(
                        market.get_position(user_id),
                        "Should exists if we are going to sell outcomes.",
                    )

                    current_position_value_usd = current_position.amounts_current[
                        trade.outcome
                    ]
                    amount_to_sell: USD | OutcomeToken
                    if current_position_value_usd <= trade.amount:
                        logger.warning(
                            f"Current value of position {trade.outcome=}, {current_position_value_usd=} is less than the desired selling amount {trade.amount=}. Selling all."
                        )
                        # In case the agent asked to sell too much, provide the amount to sell as all outcome tokens, instead of in USD, to minimze fx fluctuations when selling.
                        amount_to_sell = current_position.amounts_ot[trade.outcome]
                    else:
                        amount_to_sell = trade.amount
                    id = market.sell_tokens(
                        outcome=trade.outcome,
                        amount=amount_to_sell,
                    )
                case _:
                    raise ValueError(f"Unexpected trade type {trade.trade_type}.")
            placed_trades.append(PlacedTrade.from_trade(trade, id))
        else:
            logger.info(
                f"Trade execution skipped because, {self.place_trades=} or {still_tradeable=}."
            )

    traded_market = ProcessedTradedMarket(
        answer=processed_market.answer, trades=placed_trades
    )
    logger.info(f"Traded market {market.question=} from {market.url=}.")
    return traded_market

after_process_market

after_process_market(
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_market(
    self,
    market_type: MarketType,
    market: AgentMarket,
    processed_market: ProcessedMarket | None,
) -> None:
    api_keys = APIKeys()
    super().after_process_market(
        market_type,
        market,
        processed_market,
    )
    if isinstance(processed_market, ProcessedTradedMarket):
        if self.store_trades:
            market.store_trades(processed_market, api_keys, self.agent_name)
        else:
            logger.info(
                f"Trades {processed_market.trades} not stored because {self.store_trades=}."
            )

before_process_markets

before_process_markets(market_type: MarketType) -> None

Executed before market processing loop starts.

Source code in prediction_market_agent_tooling/deploy/agent.py

def before_process_markets(self, market_type: MarketType) -> None:
    """
    Executed before market processing loop starts.
    """
    api_keys = APIKeys()
    self.check_min_required_balance_to_operate(market_type)
    market_type.market_class.redeem_winnings(api_keys)

process_markets

process_markets(market_type: MarketType) -> None

Processes bets placed by agents on a given market.

Source code in prediction_market_agent_tooling/deploy/agent.py

def process_markets(self, market_type: MarketType) -> None:
    """
    Processes bets placed by agents on a given market.
    """
    logger.info("Start processing of markets.")
    available_markets = self.get_markets(market_type)

    logger.info(
        f"Fetched {len(available_markets)=} markets to process, going to process {self.bet_on_n_markets_per_run=}."
    )
    processed = 0

    for market_idx, market in enumerate(available_markets):
        logger.info(
            f"Going to process market {market.url}: {market_idx+1} / {len(available_markets)}."
        )
        self.before_process_market(market_type, market)
        processed_market = self.process_market(market_type, market)
        self.after_process_market(market_type, market, processed_market)

        if processed_market is not None:
            processed += 1

        if processed == self.bet_on_n_markets_per_run:
            break

    logger.info(
        f"All markets processed. Successfully processed {processed}/{len(available_markets)}."
    )

after_process_markets

after_process_markets(market_type: MarketType) -> None

Executed after market processing loop ends.

Source code in prediction_market_agent_tooling/deploy/agent.py

def after_process_markets(self, market_type: MarketType) -> None:
    """
    Executed after market processing loop ends.
    """

check_min_required_balance_to_trade

check_min_required_balance_to_trade(
    market: AgentMarket,
) -> None

Source code in prediction_market_agent_tooling/deploy/agent.py

def check_min_required_balance_to_trade(self, market: AgentMarket) -> None:
    api_keys = APIKeys()

    # Get the strategy to know how much it will bet.
    strategy = self.get_betting_strategy(market)
    # Have a little bandwidth after the bet.
    min_required_balance_to_trade = strategy.maximum_possible_bet_amount * 1.01

    if market.get_trade_balance(api_keys) < min_required_balance_to_trade:
        raise OutOfFundsError(
            f"Minimum required balance {min_required_balance_to_trade} for agent {api_keys.bet_from_address=} is not met."
        )

get_total_amount_to_bet staticmethod

get_total_amount_to_bet(market: AgentMarket) -> USD

Source code in prediction_market_agent_tooling/deploy/agent.py

@staticmethod
def get_total_amount_to_bet(market: AgentMarket) -> USD:
    user_id = market.get_user_id(api_keys=APIKeys())

    total_amount = market.get_in_usd(market.get_tiny_bet_amount())
    existing_position = market.get_position(user_id=user_id)
    if existing_position and existing_position.total_amount_current > USD(0):
        total_amount += existing_position.total_amount_current
    return total_amount

get_betting_strategy

get_betting_strategy(
    market: AgentMarket,
) -> BettingStrategy

Override this method to customize betting strategy of your agent.

Given the market and prediction, agent uses this method to calculate optimal outcome and bet size.

Source code in prediction_market_agent_tooling/deploy/agent.py

def get_betting_strategy(self, market: AgentMarket) -> BettingStrategy:
    """
    Override this method to customize betting strategy of your agent.

    Given the market and prediction, agent uses this method to calculate optimal outcome and bet size.
    """
    total_amount = self.get_total_amount_to_bet(market)
    return MultiCategoricalMaxAccuracyBettingStrategy(bet_amount=total_amount)

build_trades

build_trades(
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
    existing_position: ExistingPosition | None,
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/agent.py

def build_trades(
    self,
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
    existing_position: ExistingPosition | None,
) -> list[Trade]:
    strategy = self.get_betting_strategy(market=market)
    trades = strategy.calculate_trades(existing_position, answer, market)
    return trades

betting_strategy

GuaranteedLossError

Bases: RuntimeError

BettingStrategy

Bases: ABC

maximum_possible_bet_amount abstractmethod property

maximum_possible_bet_amount: USD

calculate_trades abstractmethod

calculate_trades(
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@abstractmethod
def calculate_trades(
    self,
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]:
    raise NotImplementedError("Subclass should implement this.")

build_zero_usd_amount staticmethod

build_zero_usd_amount() -> USD

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def build_zero_usd_amount() -> USD:
    return USD(0)

assert_buy_trade_wont_be_guaranteed_loss staticmethod

assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]:
    clean_trades = []
    for trade in trades:
        if trade.trade_type == TradeType.BUY:
            outcome_tokens_to_get = market.get_buy_token_amount(
                trade.amount, trade.outcome
            )

            if not outcome_tokens_to_get:
                logger.info(
                    f"Could not determine buy_token_amount for trade {trade}. Skipping trade."
                )
                continue

            outcome_tokens_to_get_in_usd = market.get_token_in_usd(
                outcome_tokens_to_get.as_token
            )

            if outcome_tokens_to_get_in_usd <= trade.amount:
                raise GuaranteedLossError(
                    f"Trade {trade=} would result in guaranteed loss by getting only {outcome_tokens_to_get=}. Halting execution."
                )

        clean_trades.append(trade)

    return clean_trades

filter_trades staticmethod

filter_trades(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def filter_trades(market: AgentMarket, trades: list[Trade]) -> list[Trade]:
    trades = BettingStrategy.assert_buy_trade_wont_be_guaranteed_loss(
        market, trades
    )
    return trades

MultiCategoricalMaxAccuracyBettingStrategy

MultiCategoricalMaxAccuracyBettingStrategy(bet_amount: USD)

Bases: BettingStrategy

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def __init__(self, bet_amount: USD):
    self.bet_amount = bet_amount

bet_amount instance-attribute

bet_amount = bet_amount

maximum_possible_bet_amount property

maximum_possible_bet_amount: USD

calculate_direction staticmethod

calculate_direction(
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
) -> OutcomeStr

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def calculate_direction(
    market: AgentMarket, answer: CategoricalProbabilisticAnswer
) -> OutcomeStr:
    # We place a bet on the most likely outcome
    most_likely_outcome = max(
        answer.probabilities.items(),
        key=lambda item: item[1],
    )[0]

    return market.market_outcome_for_probability_key(most_likely_outcome)

get_other_direction staticmethod

get_other_direction(
    outcomes: Sequence[OutcomeStr], direction: OutcomeStr
) -> OutcomeStr

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def get_other_direction(
    outcomes: Sequence[OutcomeStr], direction: OutcomeStr
) -> OutcomeStr:
    # We get the first direction which is != direction.
    other_direction = [i for i in outcomes if i.lower() != direction.lower()][0]
    if INVALID_OUTCOME_LOWERCASE_IDENTIFIER in other_direction.lower():
        raise ValueError("Invalid outcome found as opposite direction. Exitting.")
    return other_direction

calculate_trades

calculate_trades(
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]

We place bet on only one outcome.

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def calculate_trades(
    self,
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]:
    """We place bet on only one outcome."""

    outcome_to_bet_on = self.calculate_direction(market, answer)

    target_position = Position(
        market_id=market.id, amounts_current={outcome_to_bet_on: self.bet_amount}
    )
    trades = self._build_rebalance_trades_from_positions(
        existing_position=existing_position,
        target_position=target_position,
        market=market,
    )
    return trades

build_zero_usd_amount staticmethod

build_zero_usd_amount() -> USD

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def build_zero_usd_amount() -> USD:
    return USD(0)

assert_buy_trade_wont_be_guaranteed_loss staticmethod

assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]:
    clean_trades = []
    for trade in trades:
        if trade.trade_type == TradeType.BUY:
            outcome_tokens_to_get = market.get_buy_token_amount(
                trade.amount, trade.outcome
            )

            if not outcome_tokens_to_get:
                logger.info(
                    f"Could not determine buy_token_amount for trade {trade}. Skipping trade."
                )
                continue

            outcome_tokens_to_get_in_usd = market.get_token_in_usd(
                outcome_tokens_to_get.as_token
            )

            if outcome_tokens_to_get_in_usd <= trade.amount:
                raise GuaranteedLossError(
                    f"Trade {trade=} would result in guaranteed loss by getting only {outcome_tokens_to_get=}. Halting execution."
                )

        clean_trades.append(trade)

    return clean_trades

filter_trades staticmethod

filter_trades(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def filter_trades(market: AgentMarket, trades: list[Trade]) -> list[Trade]:
    trades = BettingStrategy.assert_buy_trade_wont_be_guaranteed_loss(
        market, trades
    )
    return trades

MaxExpectedValueBettingStrategy

MaxExpectedValueBettingStrategy(bet_amount: USD)

Bases: MultiCategoricalMaxAccuracyBettingStrategy

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def __init__(self, bet_amount: USD):
    self.bet_amount = bet_amount

maximum_possible_bet_amount property

maximum_possible_bet_amount: USD

bet_amount instance-attribute

bet_amount = bet_amount

calculate_direction staticmethod

calculate_direction(
    market: AgentMarket,
    answer: CategoricalProbabilisticAnswer,
) -> OutcomeStr

Returns the index of the outcome with the highest expected value.

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def calculate_direction(
    market: AgentMarket, answer: CategoricalProbabilisticAnswer
) -> OutcomeStr:
    """
    Returns the index of the outcome with the highest expected value.
    """

    missing_outcomes = set([i.lower() for i in market.outcomes]) - set(
        [i.lower() for i in market.probabilities.keys()]
    )

    if missing_outcomes:
        raise ValueError(
            f"Outcomes {missing_outcomes} not found in answer probabilities {answer.probabilities}"
        )

    best_outcome = None
    best_ev = float("-inf")
    for market_outcome in market.outcomes:
        if market.probability_for_market_outcome(market_outcome) == Probability(
            0.0
        ):
            # avoid division by 0
            continue

        ev = answer.probability_for_market_outcome(
            market_outcome
        ) / market.probability_for_market_outcome(market_outcome)
        if ev > best_ev:
            best_ev = ev
            best_outcome = market_outcome

    if best_outcome is None:
        raise ValueError(
            "Cannot determine best outcome - all market probabilities are zero"
        )

    return best_outcome

calculate_trades

calculate_trades(
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]

We place bet on only one outcome.

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def calculate_trades(
    self,
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]:
    """We place bet on only one outcome."""

    outcome_to_bet_on = self.calculate_direction(market, answer)

    target_position = Position(
        market_id=market.id, amounts_current={outcome_to_bet_on: self.bet_amount}
    )
    trades = self._build_rebalance_trades_from_positions(
        existing_position=existing_position,
        target_position=target_position,
        market=market,
    )
    return trades

build_zero_usd_amount staticmethod

build_zero_usd_amount() -> USD

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def build_zero_usd_amount() -> USD:
    return USD(0)

assert_buy_trade_wont_be_guaranteed_loss staticmethod

assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]:
    clean_trades = []
    for trade in trades:
        if trade.trade_type == TradeType.BUY:
            outcome_tokens_to_get = market.get_buy_token_amount(
                trade.amount, trade.outcome
            )

            if not outcome_tokens_to_get:
                logger.info(
                    f"Could not determine buy_token_amount for trade {trade}. Skipping trade."
                )
                continue

            outcome_tokens_to_get_in_usd = market.get_token_in_usd(
                outcome_tokens_to_get.as_token
            )

            if outcome_tokens_to_get_in_usd <= trade.amount:
                raise GuaranteedLossError(
                    f"Trade {trade=} would result in guaranteed loss by getting only {outcome_tokens_to_get=}. Halting execution."
                )

        clean_trades.append(trade)

    return clean_trades

filter_trades staticmethod

filter_trades(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def filter_trades(market: AgentMarket, trades: list[Trade]) -> list[Trade]:
    trades = BettingStrategy.assert_buy_trade_wont_be_guaranteed_loss(
        market, trades
    )
    return trades

get_other_direction staticmethod

get_other_direction(
    outcomes: Sequence[OutcomeStr], direction: OutcomeStr
) -> OutcomeStr

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def get_other_direction(
    outcomes: Sequence[OutcomeStr], direction: OutcomeStr
) -> OutcomeStr:
    # We get the first direction which is != direction.
    other_direction = [i for i in outcomes if i.lower() != direction.lower()][0]
    if INVALID_OUTCOME_LOWERCASE_IDENTIFIER in other_direction.lower():
        raise ValueError("Invalid outcome found as opposite direction. Exitting.")
    return other_direction

KellyBettingStrategy

KellyBettingStrategy(
    max_bet_amount: USD,
    max_price_impact: float | None = None,
)

Bases: BettingStrategy

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def __init__(self, max_bet_amount: USD, max_price_impact: float | None = None):
    self.max_bet_amount = max_bet_amount
    self.max_price_impact = max_price_impact

max_bet_amount instance-attribute

max_bet_amount = max_bet_amount

max_price_impact instance-attribute

max_price_impact = max_price_impact

maximum_possible_bet_amount property

maximum_possible_bet_amount: USD

get_kelly_bet staticmethod

get_kelly_bet(
    market: AgentMarket,
    max_bet_amount: USD,
    direction: OutcomeStr,
    other_direction: OutcomeStr,
    answer: CategoricalProbabilisticAnswer,
    override_p_yes: float | None = None,
) -> SimpleBet

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def get_kelly_bet(
    market: AgentMarket,
    max_bet_amount: USD,
    direction: OutcomeStr,
    other_direction: OutcomeStr,
    answer: CategoricalProbabilisticAnswer,
    override_p_yes: float | None = None,
) -> SimpleBet:
    estimated_p_yes = (
        answer.probability_for_market_outcome(direction)
        if not override_p_yes
        else override_p_yes
    )

    if not market.is_binary:
        # use Kelly simple, since Kelly full only supports 2 outcomes

        kelly_bet = get_kelly_bet_simplified(
            max_bet=market.get_usd_in_token(max_bet_amount),
            market_p_yes=market.probability_for_market_outcome(direction),
            estimated_p_yes=estimated_p_yes,
            confidence=answer.confidence,
        )
    else:
        # We consider only binary markets, since the Kelly strategy is not yet implemented
        # for markets with more than 2 outcomes (https://github.com/gnosis/prediction-market-agent-tooling/issues/671).
        direction_to_bet_pool_size = market.get_outcome_token_pool_by_outcome(
            direction
        )
        other_direction_pool_size = market.get_outcome_token_pool_by_outcome(
            other_direction
        )
        kelly_bet = get_kelly_bet_full(
            yes_outcome_pool_size=direction_to_bet_pool_size,
            no_outcome_pool_size=other_direction_pool_size,
            estimated_p_yes=estimated_p_yes,
            max_bet=market.get_usd_in_token(max_bet_amount),
            confidence=answer.confidence,
            fees=market.fees,
        )
    return kelly_bet

calculate_trades

calculate_trades(
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def calculate_trades(
    self,
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]:
    # We consider the p_yes as the direction with highest probability.
    direction = MultiCategoricalMaxAccuracyBettingStrategy.calculate_direction(
        market, answer
    )
    # We get the first direction which is != direction.
    other_direction = [i for i in market.outcomes if i != direction][0]
    if INVALID_OUTCOME_LOWERCASE_IDENTIFIER in other_direction.lower():
        raise ValueError("Invalid outcome found as opposite direction. Exitting.")

    kelly_bet = self.get_kelly_bet(
        market=market,
        max_bet_amount=self.max_bet_amount,
        direction=direction,
        other_direction=other_direction,
        answer=answer,
    )

    kelly_bet_size = kelly_bet.size
    if self.max_price_impact:
        # Adjust amount
        max_price_impact_bet_amount = self.calculate_bet_amount_for_price_impact(
            market, kelly_bet, direction=direction
        )

        # We just don't want Kelly size to extrapolate price_impact - hence we take the min.
        kelly_bet_size = min(kelly_bet.size, max_price_impact_bet_amount)

    bet_outcome = direction if kelly_bet.direction else other_direction
    amounts = {
        bet_outcome: market.get_token_in_usd(kelly_bet_size),
    }
    target_position = Position(market_id=market.id, amounts_current=amounts)
    trades = self._build_rebalance_trades_from_positions(
        existing_position, target_position, market=market
    )
    return trades

calculate_price_impact_for_bet_amount

calculate_price_impact_for_bet_amount(
    outcome_idx: int,
    bet_amount: CollateralToken,
    pool_balances: list[OutcomeWei],
    fees: MarketFees,
) -> float

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def calculate_price_impact_for_bet_amount(
    self,
    outcome_idx: int,
    bet_amount: CollateralToken,
    pool_balances: list[OutcomeWei],
    fees: MarketFees,
) -> float:
    prices = AgentMarket.compute_fpmm_probabilities(pool_balances)
    expected_price = prices[outcome_idx]

    tokens_to_buy = get_buy_outcome_token_amount(
        bet_amount, outcome_idx, [i.as_outcome_token for i in pool_balances], fees
    )

    actual_price = bet_amount.value / tokens_to_buy.value
    # price_impact should always be > 0
    price_impact = (actual_price - expected_price) / expected_price
    return price_impact

calculate_bet_amount_for_price_impact

calculate_bet_amount_for_price_impact(
    market: AgentMarket,
    kelly_bet: SimpleBet,
    direction: OutcomeStr,
) -> CollateralToken

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def calculate_bet_amount_for_price_impact(
    self, market: AgentMarket, kelly_bet: SimpleBet, direction: OutcomeStr
) -> CollateralToken:
    def calculate_price_impact_deviation_from_target_price_impact(
        bet_amount_usd: float,  # Needs to be float because it's used in minimize_scalar internally.
    ) -> float:
        outcome_idx = market.get_outcome_index(direction)
        price_impact = self.calculate_price_impact_for_bet_amount(
            outcome_idx=outcome_idx,
            bet_amount=market.get_usd_in_token(USD(bet_amount_usd)),
            pool_balances=pool_balances,
            fees=market.fees,
        )
        # We return abs for the algorithm to converge to 0 instead of the min (and possibly negative) value.

        max_price_impact = check_not_none(self.max_price_impact)
        return abs(price_impact - max_price_impact)

    if not market.outcome_token_pool:
        logger.warning(
            "Market outcome_token_pool is None, cannot calculate bet amount"
        )
        return kelly_bet.size

    pool_balances = [i.as_outcome_wei for i in market.outcome_token_pool.values()]
    # stay float for compatibility with `minimize_scalar`
    total_pool_balance = sum([i.value for i in market.outcome_token_pool.values()])

    # The bounds below have been found to work heuristically.
    optimized_bet_amount = minimize_scalar(
        calculate_price_impact_deviation_from_target_price_impact,
        bounds=(0, 1000 * total_pool_balance),
        method="bounded",
        tol=1e-11,
        options={"maxiter": 10000},
    )
    return CollateralToken(optimized_bet_amount.x)

build_zero_usd_amount staticmethod

build_zero_usd_amount() -> USD

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def build_zero_usd_amount() -> USD:
    return USD(0)

assert_buy_trade_wont_be_guaranteed_loss staticmethod

assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]:
    clean_trades = []
    for trade in trades:
        if trade.trade_type == TradeType.BUY:
            outcome_tokens_to_get = market.get_buy_token_amount(
                trade.amount, trade.outcome
            )

            if not outcome_tokens_to_get:
                logger.info(
                    f"Could not determine buy_token_amount for trade {trade}. Skipping trade."
                )
                continue

            outcome_tokens_to_get_in_usd = market.get_token_in_usd(
                outcome_tokens_to_get.as_token
            )

            if outcome_tokens_to_get_in_usd <= trade.amount:
                raise GuaranteedLossError(
                    f"Trade {trade=} would result in guaranteed loss by getting only {outcome_tokens_to_get=}. Halting execution."
                )

        clean_trades.append(trade)

    return clean_trades

filter_trades staticmethod

filter_trades(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def filter_trades(market: AgentMarket, trades: list[Trade]) -> list[Trade]:
    trades = BettingStrategy.assert_buy_trade_wont_be_guaranteed_loss(
        market, trades
    )
    return trades

MaxAccuracyWithKellyScaledBetsStrategy

MaxAccuracyWithKellyScaledBetsStrategy(max_bet_amount: USD)

Bases: BettingStrategy

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def __init__(self, max_bet_amount: USD):
    self.max_bet_amount = max_bet_amount

max_bet_amount instance-attribute

max_bet_amount = max_bet_amount

maximum_possible_bet_amount property

maximum_possible_bet_amount: USD

adjust_bet_amount

adjust_bet_amount(
    existing_position: ExistingPosition | None,
    market: AgentMarket,
) -> USD

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def adjust_bet_amount(
    self, existing_position: ExistingPosition | None, market: AgentMarket
) -> USD:
    existing_position_total_amount = (
        existing_position.total_amount_current if existing_position else USD(0)
    )
    return self.max_bet_amount + existing_position_total_amount

calculate_trades

calculate_trades(
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

def calculate_trades(
    self,
    existing_position: ExistingPosition | None,
    answer: CategoricalProbabilisticAnswer,
    market: AgentMarket,
) -> list[Trade]:
    adjusted_bet_amount_usd = self.adjust_bet_amount(existing_position, market)

    outcome = get_most_probable_outcome(answer.probabilities)

    direction = MultiCategoricalMaxAccuracyBettingStrategy.calculate_direction(
        market, answer
    )
    # We get the first direction which is != direction.
    other_direction = (
        MultiCategoricalMaxAccuracyBettingStrategy.get_other_direction(
            outcomes=market.outcomes, direction=direction
        )
    )

    # We ignore the direction nudge given by Kelly, hence we assume we have a perfect prediction.
    estimated_p_yes = 1.0

    kelly_bet = KellyBettingStrategy.get_kelly_bet(
        market=market,
        max_bet_amount=adjusted_bet_amount_usd,
        direction=direction,
        other_direction=other_direction,
        answer=answer,
        override_p_yes=estimated_p_yes,
    )

    kelly_bet_size_usd = market.get_token_in_usd(kelly_bet.size)

    amounts = {
        outcome: kelly_bet_size_usd,
    }
    target_position = Position(market_id=market.id, amounts_current=amounts)

    trades = self._build_rebalance_trades_from_positions(
        existing_position=existing_position,
        target_position=target_position,
        market=market,
    )
    return trades

build_zero_usd_amount staticmethod

build_zero_usd_amount() -> USD

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def build_zero_usd_amount() -> USD:
    return USD(0)

assert_buy_trade_wont_be_guaranteed_loss staticmethod

assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def assert_buy_trade_wont_be_guaranteed_loss(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]:
    clean_trades = []
    for trade in trades:
        if trade.trade_type == TradeType.BUY:
            outcome_tokens_to_get = market.get_buy_token_amount(
                trade.amount, trade.outcome
            )

            if not outcome_tokens_to_get:
                logger.info(
                    f"Could not determine buy_token_amount for trade {trade}. Skipping trade."
                )
                continue

            outcome_tokens_to_get_in_usd = market.get_token_in_usd(
                outcome_tokens_to_get.as_token
            )

            if outcome_tokens_to_get_in_usd <= trade.amount:
                raise GuaranteedLossError(
                    f"Trade {trade=} would result in guaranteed loss by getting only {outcome_tokens_to_get=}. Halting execution."
                )

        clean_trades.append(trade)

    return clean_trades

filter_trades staticmethod

filter_trades(
    market: AgentMarket, trades: list[Trade]
) -> list[Trade]

Source code in prediction_market_agent_tooling/deploy/betting_strategy.py

@staticmethod
def filter_trades(market: AgentMarket, trades: list[Trade]) -> list[Trade]:
    trades = BettingStrategy.assert_buy_trade_wont_be_guaranteed_loss(
        market, trades
    )
    return trades

constants

MARKET_TYPE_KEY module-attribute

MARKET_TYPE_KEY = 'market_type'

REPOSITORY_KEY module-attribute

REPOSITORY_KEY = 'repository'

OWNER_KEY module-attribute

OWNER_KEY = 'owner'

INVALID_OUTCOME_LOWERCASE_IDENTIFIER module-attribute

INVALID_OUTCOME_LOWERCASE_IDENTIFIER = 'invalid'

YES_OUTCOME_LOWERCASE_IDENTIFIER module-attribute

YES_OUTCOME_LOWERCASE_IDENTIFIER = 'yes'

NO_OUTCOME_LOWERCASE_IDENTIFIER module-attribute

NO_OUTCOME_LOWERCASE_IDENTIFIER = 'no'

gcp

deploy

deploy_to_gcp

deploy_to_gcp(
    gcp_fname: str,
    function_file: str,
    requirements_file: Optional[str],
    extra_deps: list[str],
    labels: dict[str, str] | None,
    env_vars: dict[str, str] | None,
    secrets: dict[str, str] | None,
    memory: int,
    entrypoint_function_name: str,
    timeout: int,
) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/deploy.py

def deploy_to_gcp(
    gcp_fname: str,
    function_file: str,
    requirements_file: t.Optional[str],
    extra_deps: list[str],
    labels: dict[str, str] | None,
    env_vars: dict[str, str] | None,
    secrets: dict[str, str] | None,
    memory: int,  # in MB
    entrypoint_function_name: str,
    timeout: int,
) -> str:
    if requirements_file and not os.path.exists(requirements_file):
        raise ValueError(f"File {requirements_file} does not exist")

    if not os.path.exists(function_file):
        raise ValueError(f"File {function_file} does not exist")

    # Make a tempdir to store the requirements file and the function
    with tempfile.TemporaryDirectory() as tempdir:
        # Copy function_file to tempdir/main.py
        shutil.copy(function_file, f"{tempdir}/main.py")

        # If the file is a .toml file, convert it to a requirements.txt file
        if requirements_file is None:
            # Just create an empty file.
            with open(f"{tempdir}/requirements.txt", "w"):
                pass
        elif requirements_file.endswith(".toml"):
            export_requirements_from_toml(output_dir=tempdir)
        else:
            shutil.copy(requirements_file, f"{tempdir}/requirements.txt")

        if extra_deps:
            with open(f"{tempdir}/requirements.txt", "a") as f:
                for dep in extra_deps:
                    f.write(f"{dep}\n")

        # Create the topic used to trigger the function. Note we use the
        # convention that the topic name is the same as the function name
        subprocess.run(gcloud_create_topic_cmd(gcp_fname), shell=True, check=True)

        # Deploy the function
        cmd = gcloud_deploy_cmd(
            gcp_function_name=gcp_fname,
            source=tempdir,
            entry_point=entrypoint_function_name,
            labels=labels,
            env_vars=env_vars,
            secrets=secrets,
            memory=memory,
            timeout=timeout,
        )
        try:
            subprocess.run(cmd, shell=True, check=True)
        except Exception:
            # Delete previously created topic if we fail to deploy the function and reraise.
            subprocess.run(gcloud_delete_topic_cmd(gcp_fname), shell=True, check=True)
            raise

            # TODO test the deployment without placing a bet

    return gcp_fname

schedule_deployed_gcp_function

schedule_deployed_gcp_function(
    function_name: str, cron_schedule: str
) -> None

Source code in prediction_market_agent_tooling/deploy/gcp/deploy.py

def schedule_deployed_gcp_function(function_name: str, cron_schedule: str) -> None:
    # Validate the cron schedule
    if not CronValidator().parse(cron_schedule):
        raise ValueError(f"Invalid cron schedule {cron_schedule}")

    cmd = gcloud_schedule_cmd(function_name=function_name, cron_schedule=cron_schedule)
    subprocess.run(cmd, shell=True, check=True)

run_deployed_gcp_function

run_deployed_gcp_function(
    function_name: str,
) -> requests.Response

Source code in prediction_market_agent_tooling/deploy/gcp/deploy.py

def run_deployed_gcp_function(function_name: str) -> requests.Response:
    uri = get_gcloud_function_uri(function_name)
    header = {"Authorization": f"Bearer {get_gcloud_id_token()}"}
    return requests.post(uri, headers=header)

remove_deployed_gcp_function

remove_deployed_gcp_function(function_name: str) -> None

Source code in prediction_market_agent_tooling/deploy/gcp/deploy.py

def remove_deployed_gcp_function(function_name: str) -> None:
    if function_name in get_gcloud_topics():
        subprocess.run(gcloud_delete_topic_cmd(function_name), shell=True, check=True)
    subprocess.run(gcloud_delete_function_cmd(function_name), shell=True, check=True)

kubernetes_models

Metadata

Bases: BaseModel

creationTimestamp instance-attribute

creationTimestamp: int

generation instance-attribute

generation: int

name instance-attribute

name: str

namespace instance-attribute

namespace: str

resourceVersion instance-attribute

resourceVersion: str

uid instance-attribute

uid: str

labels instance-attribute

labels: dict[str, str]

Metadata1

Bases: BaseModel

creationTimestamp instance-attribute

creationTimestamp: int | None

name instance-attribute

name: str

Metadata2

Bases: BaseModel

creationTimestamp instance-attribute

creationTimestamp: int | None

name instance-attribute

name: str

SecretKeyRef

Bases: BaseModel

key instance-attribute

key: str

name instance-attribute

name: str

optional instance-attribute

optional: bool

ValueFrom

Bases: BaseModel

secretKeyRef instance-attribute

secretKeyRef: SecretKeyRef

EnvItem

Bases: BaseModel

name instance-attribute

name: str

valueFrom instance-attribute

valueFrom: ValueFrom

ConfigMapRef

Bases: BaseModel

name instance-attribute

name: str

optional instance-attribute

optional: bool

EnvFromItem

Bases: BaseModel

configMapRef instance-attribute

configMapRef: ConfigMapRef

Limits

Bases: BaseModel

cpu instance-attribute

cpu: str

memory instance-attribute

memory: str

Requests

Bases: BaseModel

cpu instance-attribute

cpu: str

memory instance-attribute

memory: str

Resources

Bases: BaseModel

limits instance-attribute

limits: Limits

requests instance-attribute

requests: Requests

Container

Bases: BaseModel

env instance-attribute

env: list[EnvItem]

envFrom instance-attribute

envFrom: list[EnvFromItem]

image instance-attribute

image: str

imagePullPolicy instance-attribute

imagePullPolicy: str

name instance-attribute

name: str

resources instance-attribute

resources: Resources

terminationMessagePath instance-attribute

terminationMessagePath: str

terminationMessagePolicy instance-attribute

terminationMessagePolicy: str

NodeSelector

Bases: BaseModel

role instance-attribute

role: str

Toleration

Bases: BaseModel

effect instance-attribute

effect: str

key instance-attribute

key: str

operator instance-attribute

operator: str

value instance-attribute

value: str

Spec2

Bases: BaseModel

automountServiceAccountToken instance-attribute

automountServiceAccountToken: bool

containers instance-attribute

containers: list[Container]

dnsPolicy instance-attribute

dnsPolicy: str

enableServiceLinks instance-attribute

enableServiceLinks: bool

nodeSelector instance-attribute

nodeSelector: NodeSelector

restartPolicy instance-attribute

restartPolicy: str

schedulerName instance-attribute

schedulerName: str

securityContext instance-attribute

securityContext: dict[str, Any]

shareProcessNamespace instance-attribute

shareProcessNamespace: bool

terminationGracePeriodSeconds instance-attribute

terminationGracePeriodSeconds: int

tolerations instance-attribute

tolerations: list[Toleration]

Template

Bases: BaseModel

metadata instance-attribute

metadata: Metadata2

spec instance-attribute

spec: Spec2

Spec1

Bases: BaseModel

activeDeadlineSeconds instance-attribute

activeDeadlineSeconds: int

backoffLimit instance-attribute

backoffLimit: int

completions instance-attribute

completions: int

manualSelector instance-attribute

manualSelector: bool

parallelism instance-attribute

parallelism: int

template instance-attribute

template: Template

JobTemplate

Bases: BaseModel

metadata instance-attribute

metadata: Metadata1

spec instance-attribute

spec: Spec1

Spec

Bases: BaseModel

concurrencyPolicy instance-attribute

concurrencyPolicy: str

failedJobsHistoryLimit instance-attribute

failedJobsHistoryLimit: int

jobTemplate instance-attribute

jobTemplate: JobTemplate

schedule instance-attribute

schedule: str

successfulJobsHistoryLimit instance-attribute

successfulJobsHistoryLimit: int

suspend instance-attribute

suspend: bool

timeZone instance-attribute

timeZone: str

Status

Bases: BaseModel

lastScheduleTime class-attribute instance-attribute

lastScheduleTime: str | None = None

lastSuccessfulTime class-attribute instance-attribute

lastSuccessfulTime: str | None = None

KubernetesCronJob

Bases: BaseModel

apiVersion instance-attribute

apiVersion: str

kind instance-attribute

kind: str

metadata instance-attribute

metadata: Metadata

spec instance-attribute

spec: Spec

status instance-attribute

status: Status

Metadata3

Bases: BaseModel

resourceVersion instance-attribute

resourceVersion: str

KubernetesCronJobsModel

Bases: BaseModel

apiVersion instance-attribute

apiVersion: str

items instance-attribute

items: list[KubernetesCronJob]

kind instance-attribute

kind: str

metadata instance-attribute

metadata: Metadata3

utils

gcloud_deploy_cmd

gcloud_deploy_cmd(
    gcp_function_name: str,
    source: str,
    entry_point: str,
    labels: dict[str, str] | None,
    env_vars: dict[str, str] | None,
    secrets: dict[str, str] | None,
    memory: int,
    timeout: int = 180,
    retry_on_failure: bool = False,
) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcloud_deploy_cmd(
    gcp_function_name: str,
    source: str,
    entry_point: str,
    labels: dict[str, str] | None,
    env_vars: dict[str, str] | None,
    secrets: dict[str, str] | None,
    memory: int,  # in MB
    timeout: int = 180,
    retry_on_failure: bool = False,
) -> str:
    cmd = (
        f"gcloud functions deploy {gcp_function_name} "
        f"--runtime {get_gcloud_python_runtime_str()} "
        f"--trigger-topic {gcp_function_name} "
        f"--gen2 "
        f"--region {get_gcloud_region()} "
        f"--source {source} "
        f"--entry-point {entry_point} "
        f"--memory {memory}MB "
        f"--no-allow-unauthenticated "
        f"--timeout {timeout}s "
        # Explicitly set no concurrency, min instances to 0 (agent is executed only once in a while) and max instances to 1 (parallel agents aren't allowed).
        "--concurrency 1 "
        "--min-instances 0 "
        "--max-instances 1 "
    )
    if retry_on_failure:
        cmd += "--retry "
    if labels:
        for k, v in labels.items():
            cmd += f'--update-labels {k}="{v}" '
    if env_vars:
        for k, v in env_vars.items():
            cmd += f'--set-env-vars {k}="{v}" '
    if secrets:
        for k, v in secrets.items():
            cmd += f'--set-secrets {k}="{v}" '

    return cmd

gcloud_schedule_cmd

gcloud_schedule_cmd(
    function_name: str, cron_schedule: str
) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcloud_schedule_cmd(function_name: str, cron_schedule: str) -> str:
    return (
        f"gcloud scheduler jobs create pubsub {function_name} "
        f"--schedule '{cron_schedule}' "
        f"--topic {function_name} "
        f"--location {get_gcloud_region()} "
        "--message-body '{}' "
    )

gcloud_delete_function_cmd

gcloud_delete_function_cmd(fname: str) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcloud_delete_function_cmd(fname: str) -> str:
    return f"gcloud functions delete {fname} --region={get_gcloud_region()} --quiet"

gcloud_get_topics_cmd

gcloud_get_topics_cmd() -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcloud_get_topics_cmd() -> str:
    return "gcloud pubsub topics list --format='value(name)' | awk -F'/' '{print $NF}'"

get_gcloud_topics

get_gcloud_topics() -> list[str]

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_topics() -> list[str]:
    return (
        subprocess.run(
            gcloud_get_topics_cmd(),
            shell=True,
            capture_output=True,
            check=True,
        )
        .stdout.decode()
        .strip()
        .split("\n")
    )

gcloud_create_topic_cmd

gcloud_create_topic_cmd(topic_name: str) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcloud_create_topic_cmd(topic_name: str) -> str:
    return f"gcloud pubsub topics create {topic_name}"

gcloud_delete_topic_cmd

gcloud_delete_topic_cmd(topic_name: str) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcloud_delete_topic_cmd(topic_name: str) -> str:
    return f"gcloud pubsub topics delete {topic_name}"

get_gcloud_project_id

get_gcloud_project_id() -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_project_id() -> str:
    return (
        subprocess.run(
            "gcloud config get-value project",
            shell=True,
            capture_output=True,
            check=True,
        )
        .stdout.decode()
        .strip()
    )

get_gcloud_parent

get_gcloud_parent() -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_parent() -> str:
    return f"projects/{get_gcloud_project_id()}/locations/{get_gcloud_region()}"

get_gcloud_id_token

get_gcloud_id_token() -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_id_token() -> str:
    return (
        subprocess.run(
            "gcloud auth print-identity-token",
            shell=True,
            capture_output=True,
            check=True,
        )
        .stdout.decode()
        .strip()
    )

get_gcloud_region

get_gcloud_region() -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_region() -> str:
    return "europe-west2"  # London

get_gcloud_python_runtime_str

get_gcloud_python_runtime_str() -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_python_runtime_str() -> str:
    return f"python{sys.version_info.major}{sys.version_info.minor}"

get_gcloud_function_uri

get_gcloud_function_uri(function_name: str) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcloud_function_uri(function_name: str) -> str:
    return (
        subprocess.run(
            f"gcloud functions describe {function_name} --region {get_gcloud_region()} --format='value(url)'",
            shell=True,
            capture_output=True,
            check=True,
        )
        .stdout.decode()
        .strip()
    )

api_keys_to_str

api_keys_to_str(api_keys: dict[str, str]) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def api_keys_to_str(api_keys: dict[str, str]) -> str:
    return " ".join([f"{k}={v}" for k, v in api_keys.items()])

list_gcp_functions

list_gcp_functions() -> list[Function]

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def list_gcp_functions() -> list[Function]:
    client = FunctionServiceClient()
    functions = list(client.list_functions(parent=get_gcloud_parent()))
    return functions

list_gcp_cronjobs

list_gcp_cronjobs(
    namespace: str,
) -> KubernetesCronJobsModel

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def list_gcp_cronjobs(namespace: str) -> KubernetesCronJobsModel:
    return KubernetesCronJobsModel.model_validate_json(
        subprocess.run(
            f"kubectl get cronjobs -o json -n {namespace}",
            shell=True,
            capture_output=True,
            check=True,
        )
        .stdout.decode()
        .strip()
    )

get_gcp_configmap_data

get_gcp_configmap_data(
    namespace: str, name: str
) -> dict[str, str]

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcp_configmap_data(namespace: str, name: str) -> dict[str, str]:
    data: dict[str, str] = json.loads(
        subprocess.run(
            f"kubectl get configmap {name} -o json -n {namespace}",
            shell=True,
            capture_output=True,
            check=True,
        )
        .stdout.decode()
        .strip()
    )["data"]
    return data

get_gcp_function

get_gcp_function(fname: str) -> Function

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def get_gcp_function(fname: str) -> Function:
    response = list_gcp_functions()
    for function in response:
        if function.name.split("/")[-1] == fname:
            return function

    fnames = [f.name.split("/")[-1] for f in response]
    raise ValueError(f"Function {fname} not found in function list {fnames}")

gcp_function_is_active

gcp_function_is_active(fname: str) -> bool

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

def gcp_function_is_active(fname: str) -> bool:
    return get_gcp_function(fname).state == Function.State.ACTIVE

gcp_get_secret_value cached

gcp_get_secret_value(
    name: str, version: str = "latest"
) -> str

Source code in prediction_market_agent_tooling/deploy/gcp/utils.py

@cache
def gcp_get_secret_value(name: str, version: str = "latest") -> str:
    client = SecretManagerServiceClient()
    return client.access_secret_version(
        name=f"projects/{get_gcloud_project_id()}/secrets/{name}/versions/{version}"
    ).payload.data.decode("utf-8")

trade_interval

TradeInterval

Bases: ABC

get abstractmethod

get(market: AgentMarket) -> timedelta

Source code in prediction_market_agent_tooling/deploy/trade_interval.py

@abstractmethod
def get(
    self,
    market: AgentMarket,
) -> timedelta:
    raise NotImplementedError("Subclass should implement this.")

FixedInterval

FixedInterval(interval: timedelta)

Bases: TradeInterval

For trades at a fixed interval.

Source code in prediction_market_agent_tooling/deploy/trade_interval.py

def __init__(self, interval: timedelta):
    self.interval = interval

interval instance-attribute

interval = interval

get

get(market: AgentMarket) -> timedelta

Source code in prediction_market_agent_tooling/deploy/trade_interval.py

def get(
    self,
    market: AgentMarket,
) -> timedelta:
    return self.interval

MarketLifetimeProportionalInterval

MarketLifetimeProportionalInterval(max_trades: int)

Bases: TradeInterval

For uniformly distributed trades over the market's lifetime.

Source code in prediction_market_agent_tooling/deploy/trade_interval.py

def __init__(self, max_trades: int):
    self.max_trades = max_trades

max_trades instance-attribute

max_trades = max_trades

get

get(market: AgentMarket) -> timedelta

Source code in prediction_market_agent_tooling/deploy/trade_interval.py

def get(
    self,
    market: AgentMarket,
) -> timedelta:
    created_time = check_not_none(market.created_time)
    close_time = check_not_none(market.close_time)
    return (close_time - created_time) / self.max_trades