Skip to content

Benchmark Module

prediction_market_agent_tooling.benchmark

agents

AbstractBenchmarkedAgent 

AbstractBenchmarkedAgent(
    agent_name: str,
    max_workers: Optional[int] = None,
    model: str | None = None,
)
Source code in prediction_market_agent_tooling/benchmark/agents.py 
15
16
17
18
19
20
21
22
23
def __init__(
    self,
    agent_name: str,
    max_workers: t.Optional[int] = None,
    model: str | None = None,
):
    self.model = model
    self.agent_name = agent_name
    self.max_workers = max_workers  # Limit the number of workers that can run this worker in parallel threads
model instance-attribute 
model = model
agent_name instance-attribute 
agent_name = agent_name
max_workers instance-attribute 
max_workers = max_workers
is_predictable 
is_predictable(market_question: str) -> bool

Override if the agent can decide to not predict the question, before doing the hard work.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
25
26
27
28
29
def is_predictable(self, market_question: str) -> bool:
    """
    Override if the agent can decide to not predict the question, before doing the hard work.
    """
    return True
predict 
predict(market_question: str) -> Prediction

Predict the outcome of the market question.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
31
32
33
34
35
def predict(self, market_question: str) -> Prediction:
    """
    Predict the outcome of the market question.
    """
    raise NotImplementedError
check_and_predict 
check_and_predict(market_question: str) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
37
38
39
40
41
def check_and_predict(self, market_question: str) -> Prediction:
    is_predictable = self.is_predictable(market_question=market_question)
    if not is_predictable:
        return Prediction(is_predictable=is_predictable)
    return self.predict(market_question=market_question)
is_predictable_restricted 
is_predictable_restricted(
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> bool

Override if the agent can decide to not predict the question, before doing the hard work.

Data used for the evaluation must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
43
44
45
46
47
48
49
50
51
52
53
def is_predictable_restricted(
    self,
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> bool:
    """
    Override if the agent can decide to not predict the question, before doing the hard work.

    Data used for the evaluation must be restricted to the time_restriction_up_to.
    """
    return True
predict_restricted 
predict_restricted(
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction

Predict the outcome of the market question.

Data used for the prediction must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
55
56
57
58
59
60
61
62
63
64
65
def predict_restricted(
    self,
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction:
    """
    Predict the outcome of the market question.

    Data used for the prediction must be restricted to the time_restriction_up_to.
    """
    raise NotImplementedError
check_and_predict_restricted 
check_and_predict_restricted(
    market: AgentMarket, time_restriction_up_to: DatetimeUTC
) -> Prediction

Data used must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def check_and_predict_restricted(
    self,
    market: AgentMarket,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction:
    """
    Data used must be restricted to the time_restriction_up_to.
    """
    is_predictable = self.is_predictable_restricted(
        market_question=market.question,
        time_restriction_up_to=time_restriction_up_to,
    )
    if not is_predictable:
        return Prediction(is_predictable=is_predictable)
    return self.predict_restricted(
        market_question=market.question,
        time_restriction_up_to=time_restriction_up_to,
    )

RandomAgent 

RandomAgent(
    agent_name: str,
    max_workers: Optional[int] = None,
    model: str | None = None,
)

Bases: AbstractBenchmarkedAgent

Source code in prediction_market_agent_tooling/benchmark/agents.py 
15
16
17
18
19
20
21
22
23
def __init__(
    self,
    agent_name: str,
    max_workers: t.Optional[int] = None,
    model: str | None = None,
):
    self.model = model
    self.agent_name = agent_name
    self.max_workers = max_workers  # Limit the number of workers that can run this worker in parallel threads
model instance-attribute 
model = model
agent_name instance-attribute 
agent_name = agent_name
max_workers instance-attribute 
max_workers = max_workers
predict 
predict(market_question: str) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
88
89
90
91
92
93
94
95
def predict(self, market_question: str) -> Prediction:
    p_yes, confidence = random.random(), random.random()

    return Prediction(
        outcome_prediction=CategoricalProbabilisticAnswer.from_probabilistic_answer(
            ProbabilisticAnswer(p_yes=Probability(p_yes), confidence=confidence),
        ),
    )
predict_restricted 
predict_restricted(
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
 97
 98
 99
100
def predict_restricted(
    self, market_question: str, time_restriction_up_to: DatetimeUTC
) -> Prediction:
    return self.predict(market_question)
is_predictable 
is_predictable(market_question: str) -> bool

Override if the agent can decide to not predict the question, before doing the hard work.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
25
26
27
28
29
def is_predictable(self, market_question: str) -> bool:
    """
    Override if the agent can decide to not predict the question, before doing the hard work.
    """
    return True
check_and_predict 
check_and_predict(market_question: str) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
37
38
39
40
41
def check_and_predict(self, market_question: str) -> Prediction:
    is_predictable = self.is_predictable(market_question=market_question)
    if not is_predictable:
        return Prediction(is_predictable=is_predictable)
    return self.predict(market_question=market_question)
is_predictable_restricted 
is_predictable_restricted(
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> bool

Override if the agent can decide to not predict the question, before doing the hard work.

Data used for the evaluation must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
43
44
45
46
47
48
49
50
51
52
53
def is_predictable_restricted(
    self,
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> bool:
    """
    Override if the agent can decide to not predict the question, before doing the hard work.

    Data used for the evaluation must be restricted to the time_restriction_up_to.
    """
    return True
check_and_predict_restricted 
check_and_predict_restricted(
    market: AgentMarket, time_restriction_up_to: DatetimeUTC
) -> Prediction

Data used must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def check_and_predict_restricted(
    self,
    market: AgentMarket,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction:
    """
    Data used must be restricted to the time_restriction_up_to.
    """
    is_predictable = self.is_predictable_restricted(
        market_question=market.question,
        time_restriction_up_to=time_restriction_up_to,
    )
    if not is_predictable:
        return Prediction(is_predictable=is_predictable)
    return self.predict_restricted(
        market_question=market.question,
        time_restriction_up_to=time_restriction_up_to,
    )

FixedAgent 

FixedAgent(
    fixed_answer: bool,
    agent_name: str,
    max_workers: int | None = None,
)

Bases: AbstractBenchmarkedAgent

Source code in prediction_market_agent_tooling/benchmark/agents.py 
104
105
106
107
108
def __init__(
    self, fixed_answer: bool, agent_name: str, max_workers: int | None = None
):
    super().__init__(agent_name, max_workers)
    self.fixed_answer = fixed_answer
fixed_answer instance-attribute 
fixed_answer = fixed_answer
model instance-attribute 
model = model
agent_name instance-attribute 
agent_name = agent_name
max_workers instance-attribute 
max_workers = max_workers
predict 
predict(market_question: str) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
110
111
112
113
114
115
116
117
118
119
120
def predict(self, market_question: str) -> Prediction:
    p_yes, confidence = 1.0 if self.fixed_answer else 0.0, 1.0

    return Prediction(
        outcome_prediction=CategoricalProbabilisticAnswer.from_probabilistic_answer(
            ProbabilisticAnswer(
                p_yes=Probability(p_yes),
                confidence=confidence,
            )
        )
    )
predict_restricted 
predict_restricted(
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
122
123
124
125
def predict_restricted(
    self, market_question: str, time_restriction_up_to: DatetimeUTC
) -> Prediction:
    return self.predict(market_question)
is_predictable 
is_predictable(market_question: str) -> bool

Override if the agent can decide to not predict the question, before doing the hard work.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
25
26
27
28
29
def is_predictable(self, market_question: str) -> bool:
    """
    Override if the agent can decide to not predict the question, before doing the hard work.
    """
    return True
check_and_predict 
check_and_predict(market_question: str) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/agents.py 
37
38
39
40
41
def check_and_predict(self, market_question: str) -> Prediction:
    is_predictable = self.is_predictable(market_question=market_question)
    if not is_predictable:
        return Prediction(is_predictable=is_predictable)
    return self.predict(market_question=market_question)
is_predictable_restricted 
is_predictable_restricted(
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> bool

Override if the agent can decide to not predict the question, before doing the hard work.

Data used for the evaluation must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
43
44
45
46
47
48
49
50
51
52
53
def is_predictable_restricted(
    self,
    market_question: str,
    time_restriction_up_to: DatetimeUTC,
) -> bool:
    """
    Override if the agent can decide to not predict the question, before doing the hard work.

    Data used for the evaluation must be restricted to the time_restriction_up_to.
    """
    return True
check_and_predict_restricted 
check_and_predict_restricted(
    market: AgentMarket, time_restriction_up_to: DatetimeUTC
) -> Prediction

Data used must be restricted to the time_restriction_up_to.

Source code in prediction_market_agent_tooling/benchmark/agents.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def check_and_predict_restricted(
    self,
    market: AgentMarket,
    time_restriction_up_to: DatetimeUTC,
) -> Prediction:
    """
    Data used must be restricted to the time_restriction_up_to.
    """
    is_predictable = self.is_predictable_restricted(
        market_question=market.question,
        time_restriction_up_to=time_restriction_up_to,
    )
    if not is_predictable:
        return Prediction(is_predictable=is_predictable)
    return self.predict_restricted(
        market_question=market.question,
        time_restriction_up_to=time_restriction_up_to,
    )

benchmark

Benchmarker 

Benchmarker(
    markets: Sequence[AgentMarket],
    agents: List[AbstractBenchmarkedAgent],
    metric_fns: Dict[
        str,
        Callable[
            [list[Prediction], Sequence[AgentMarket]],
            str | float | None,
        ],
    ] = {},
    cache_path: Optional[str] = None,
    only_cached: bool = False,
)
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(
    self,
    markets: t.Sequence[AgentMarket],
    agents: t.List[AbstractBenchmarkedAgent],
    metric_fns: t.Dict[
        str,
        t.Callable[[list[Prediction], t.Sequence[AgentMarket]], str | float | None],
    ] = {},
    cache_path: t.Optional[str] = None,
    only_cached: bool = False,
):
    self.registered_agents: t.List[AbstractBenchmarkedAgent] = agents
    if len(set(a.agent_name for a in self.registered_agents)) != len(
        self.registered_agents
    ):
        raise ValueError("Agents must have unique names")
    if any(m.has_unsuccessful_resolution() for m in markets):
        raise ValueError(
            "Unsuccessful markets shouldn't be used in the benchmark, please filter them out."
        )

    # Predictions
    self.cache_path = cache_path
    if self.cache_path and os.path.exists(self.cache_path):
        self.predictions = PredictionsCache.load(path=self.cache_path)
    else:
        self.predictions = PredictionsCache(predictions={})

    self.only_cached = only_cached
    self.markets: t.Sequence[AgentMarket] = (
        [
            m
            for m in markets
            if all(
                self.predictions.has_market(
                    agent_name=agent.agent_name, question=m.question
                )
                for agent in self.registered_agents
            )
        ]
        if self.only_cached
        else markets
    )

    # Metrics
    self.metric_fns = metric_fns
    predefined_metric_fns = {
        "MSE for `p_yes`": self._compute_mse,
        "Mean confidence": self._compute_mean_confidence,
        "% within +-0.05": lambda predictions, markets: self._compute_percentage_within_range(
            predictions, markets, average_error_tolerance=0.05
        ),
        "% within +-0.1": lambda predictions, markets: self._compute_percentage_within_range(
            predictions, markets, average_error_tolerance=0.1
        ),
        "% within +-0.2": lambda predictions, markets: self._compute_percentage_within_range(
            predictions, markets, average_error_tolerance=0.2
        ),
        "% correct outcome": self._compute_correct_outcome_percentage,
        "% precision for `yes`": lambda predictions, markets: self._compute_precision_and_recall_percentages(
            predictions, markets
        )[
            0
        ],
        "% precision for `no`": lambda predictions, markets: self._compute_precision_and_recall_percentages(
            predictions, markets
        )[
            0
        ],
        "% recall for `yes`": lambda predictions, markets: self._compute_precision_and_recall_percentages(
            predictions, markets
        )[
            1
        ],
        "% recall for `no`": lambda predictions, markets: self._compute_precision_and_recall_percentages(
            predictions, markets
        )[
            1
        ],
        "confidence/p_yes error correlation": self._compute_confidence_p_yes_error_correlation,
        "Proportion answerable": self._compute_ratio_evaluated_as_answerable,
        "Proportion answered": self._compute_ratio_answered,
        "Mean cost ($)": self._compute_mean_cost,
        "Mean time (s)": self._compute_mean_time,
    }
    self.metric_fns.update(predefined_metric_fns)
registered_agents instance-attribute 
registered_agents: List[AbstractBenchmarkedAgent] = agents
cache_path instance-attribute 
cache_path = cache_path
predictions instance-attribute 
predictions = load(path=cache_path)
only_cached instance-attribute 
only_cached = only_cached
markets instance-attribute 
markets: Sequence[AgentMarket] = (
    [
        m
        for m in markets
        if all(
            has_market(
                agent_name=agent_name, question=question
            )
            for agent in registered_agents
        )
    ]
    if only_cached
    else markets
)
metric_fns instance-attribute 
metric_fns = metric_fns
add_prediction 
add_prediction(
    agent: AbstractBenchmarkedAgent,
    prediction: Prediction,
    market_question: str,
) -> None
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
110
111
112
113
114
115
116
117
118
119
120
def add_prediction(
    self,
    agent: AbstractBenchmarkedAgent,
    prediction: Prediction,
    market_question: str,
) -> None:
    self.predictions.add_prediction(
        agent_name=agent.agent_name,
        question=market_question,
        prediction=prediction,
    )
get_prediction 
get_prediction(
    agent_name: str, question: str
) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
122
123
def get_prediction(self, agent_name: str, question: str) -> Prediction:
    return self.predictions.get_prediction(agent_name=agent_name, question=question)
run_agents 
run_agents(enable_timing: bool = True) -> None
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def run_agents(self, enable_timing: bool = True) -> None:
    agent: AbstractBenchmarkedAgent  # Fix for mypy issue with tqdm.
    for agent in tqdm(self.registered_agents, desc="Running agents"):
        # Filter out cached predictions
        markets_to_run = [
            m
            for m in self.markets
            if not self.predictions.has_market(
                agent_name=agent.agent_name, question=m.question
            )
        ]

        def get_prediction_result(
            market: AgentMarket,
        ) -> tuple[str, Prediction]:
            with openai_costs(model=agent.model) as costs:
                prediction = (
                    agent.check_and_predict(market_question=market.question)
                    if not market.is_resolved()
                    else (
                        agent.check_and_predict_restricted(
                            market=market,
                            time_restriction_up_to=market.created_time,  # TODO: Add support for resolved_at and any time in between.
                        )
                        if market.created_time is not None
                        else should_not_happen()
                    )
                )
                prediction.time = costs.time
                prediction.cost = costs.cost
            return market.question, prediction

        # Run agents in parallel
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=agent.max_workers
        ) as executor:
            futures = [
                executor.submit(get_prediction_result, market)
                for market in markets_to_run
            ]
            for future in tqdm(
                concurrent.futures.as_completed(futures),
                total=len(futures),
                desc=f"Running {agent.agent_name}",
            ):
                market_question, prediction = future.result()
                self.add_prediction(
                    agent=agent,
                    prediction=prediction,
                    market_question=market_question,
                )
                if self.cache_path:
                    self.predictions.save(self.cache_path)
filter_predictions_for_answered staticmethod 
filter_predictions_for_answered(
    predictions: list[Prediction],
    markets: Sequence[AgentMarket],
) -> t.Tuple[list[Prediction], list[AgentMarket]]
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
179
180
181
182
183
184
185
186
187
188
@staticmethod
def filter_predictions_for_answered(
    predictions: list[Prediction], markets: t.Sequence[AgentMarket]
) -> t.Tuple[list[Prediction], list[AgentMarket]]:
    filtered_predictions, filtered_markets = [], []
    for p, m in zip(predictions, markets):
        if p.is_answered:
            filtered_predictions.append(p)
            filtered_markets.append(m)
    return filtered_predictions, filtered_markets
calculate_errors_between_prediction_and_market staticmethod 
calculate_errors_between_prediction_and_market(
    prediction: Prediction, market: AgentMarket
) -> list[float]
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@staticmethod
def calculate_errors_between_prediction_and_market(
    prediction: Prediction, market: AgentMarket
) -> list[float]:
    pred_probs = check_not_none(prediction.outcome_prediction).probabilities
    market_probs = market.probabilities

    # Get common outcomes between prediction and market
    common_outcomes = set(pred_probs.keys()) & set(market_probs.keys())

    errors = [
        (pred_probs[outcome] - market_probs[outcome]) for outcome in common_outcomes
    ]

    return errors
calculate_squared_errors staticmethod 
calculate_squared_errors(
    prediction: Prediction, market: AgentMarket
) -> float
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
222
223
224
225
226
227
228
@staticmethod
def calculate_squared_errors(prediction: Prediction, market: AgentMarket) -> float:
    errors = Benchmarker.calculate_errors_between_prediction_and_market(
        prediction, market
    )
    squared_errors = sum([err**2 for err in errors], 0.0)
    return squared_errors
compute_metrics 
compute_metrics() -> t.Dict[str, t.List[t.Any]]
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def compute_metrics(self) -> t.Dict[str, t.List[t.Any]]:
    metrics: dict[str, list[str | float | None]] = {}
    metrics["Agents"] = [a.agent_name for a in self.registered_agents]

    for name, fn in self.metric_fns.items():
        metrics[name] = []
        for agent in self.registered_agents:
            ordered_predictions = [
                self.get_prediction(
                    question=market.question, agent_name=agent.agent_name
                )
                for market in self.markets
            ]
            metrics[name].append(fn(ordered_predictions, self.markets))

    return metrics
get_markets_summary 
get_markets_summary() -> t.Dict[str, t.List[str | float]]
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def get_markets_summary(self) -> t.Dict[str, t.List[str | float]]:
    market_questions = [q.question for q in self.markets]
    urls = [q.url for q in self.markets]
    markets_summary: dict[str, list[str | float]] = {
        "Market Question": [
            f"[{question}]({url})" for question, url in zip(market_questions, urls)
        ],
    }

    for agent in [a.agent_name for a in self.registered_agents]:
        agent_predictions = [
            self.get_prediction(agent_name=agent, question=q)
            for q in market_questions
        ]
        markets_summary[f"{agent} p_yes"] = [
            (
                f"{p.outcome_prediction.probabilities} [{p.outcome_prediction.probable_resolution}]"
                if p.is_predictable
                and p.outcome_prediction  # Is answerable and answered
                else (
                    "S"
                    if not p.is_predictable  # Skipped (evaluated to be not predictable)
                    else (
                        "F"
                        if p.is_predictable
                        and not p.outcome_prediction  # Failed (no prediction)
                        else should_not_happen(
                            f"Unexpected case in get_markets_summary() for {p}."
                        )
                    )
                )
            )
            for p in agent_predictions
        ]
    markets_summary[f"reference probabilities"] = [
        f"{m.probabilities} [{m.probable_resolution}]" for m in self.markets
    ]
    return markets_summary
get_markets_results 
get_markets_results() -> dict[str, list[str | float]]
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
def get_markets_results(self) -> dict[str, list[str | float]]:
    outcome_counts: dict[OutcomeStr, int] = defaultdict(int)
    total_markets = len(self.markets)

    for market in self.markets:
        resolution = market.probable_resolution
        if resolution.outcome:
            outcome_counts[resolution.outcome] += 1

    proportions = {
        outcome: count / total_markets for outcome, count in outcome_counts.items()
    }
    return {
        "Number of markets": [total_markets],
        "Proportion resolved": [
            sum(1 for m in self.markets if m.is_resolved()) / total_markets
        ],
        **{
            f"Proportion {outcome}": [proportions.get(outcome, 0)]
            for outcome in outcome_counts
        },
    }
generate_markdown_report 
generate_markdown_report() -> str
Source code in prediction_market_agent_tooling/benchmark/benchmark.py 
448
449
450
451
452
453
454
455
456
457
458
459
def generate_markdown_report(self) -> str:
    md = "# Comparison Report\n\n"
    md += "## Market Results\n\n"
    md += pd.DataFrame(self.get_markets_results()).to_markdown(index=False)
    md += "\n\n"
    md += "## Agent Results\n\n"
    md += "### Summary Statistics\n\n"
    md += pd.DataFrame(self.compute_metrics()).to_markdown(index=False)
    md += "\n\n"
    md += "### Markets\n\n"
    md += pd.DataFrame(self.get_markets_summary()).to_markdown(index=False)
    return str(md)

utils

AgentPredictions module-attribute 

AgentPredictions = Dict[str, Prediction]

Predictions module-attribute 

Predictions = Dict[str, AgentPredictions]

Prediction

Bases: BaseModel

is_predictable class-attribute instance-attribute 
is_predictable: bool = True
outcome_prediction class-attribute instance-attribute 
outcome_prediction: Optional[
    CategoricalProbabilisticAnswer
] = None
time class-attribute instance-attribute 
time: Optional[float] = None
cost class-attribute instance-attribute 
cost: Optional[float] = None
is_answered property 
is_answered: bool

PredictionsCache

Bases: BaseModel

predictions instance-attribute 
predictions: Predictions
get_prediction 
get_prediction(
    agent_name: str, question: str
) -> Prediction
Source code in prediction_market_agent_tooling/benchmark/utils.py 
38
39
def get_prediction(self, agent_name: str, question: str) -> Prediction:
    return self.predictions[agent_name][question]
has_market 
has_market(agent_name: str, question: str) -> bool
Source code in prediction_market_agent_tooling/benchmark/utils.py 
41
42
43
44
def has_market(self, agent_name: str, question: str) -> bool:
    return (
        agent_name in self.predictions and question in self.predictions[agent_name]
    )
add_prediction 
add_prediction(
    agent_name: str, question: str, prediction: Prediction
) -> None
Source code in prediction_market_agent_tooling/benchmark/utils.py 
46
47
48
49
50
51
52
53
54
def add_prediction(
    self, agent_name: str, question: str, prediction: Prediction
) -> None:
    if agent_name not in self.predictions:
        self.predictions[agent_name] = {}
    assert (
        question not in self.predictions[agent_name]
    ), f"Question `{question}` already exists in the cache."
    self.predictions[agent_name][question] = prediction
save 
save(path: str) -> None
Source code in prediction_market_agent_tooling/benchmark/utils.py 
56
57
58
def save(self, path: str) -> None:
    with open(path, "w") as f:
        json.dump(self.model_dump(), f, indent=2)
load staticmethod 
load(path: str) -> PredictionsCache
Source code in prediction_market_agent_tooling/benchmark/utils.py 
60
61
62
63
@staticmethod
def load(path: str) -> "PredictionsCache":
    with open(path, "r") as f:
        return PredictionsCache.model_validate(json.load(f))

get_most_probable_outcome 

get_most_probable_outcome(
    probability_map: dict[OutcomeStr, Probability],
) -> OutcomeStr

Returns most probable outcome. If tied, returns first.

Source code in prediction_market_agent_tooling/benchmark/utils.py 
12
13
14
15
16
def get_most_probable_outcome(
    probability_map: dict[OutcomeStr, Probability],
) -> OutcomeStr:
    """Returns most probable outcome. If tied, returns first."""
    return max(probability_map, key=lambda k: float(probability_map[k]))

get_llm_api_call_cost 

get_llm_api_call_cost(
    model: str, prompt_tokens: int, completion_tokens: float
) -> float

In older versions of langchain, the cost calculation doesn't work for newer models. This is a temporary workaround to get the cost.

See: https://github.com/langchain-ai/langchain/issues/12994

Costs are in USD, per 1000 tokens.

Source code in prediction_market_agent_tooling/benchmark/utils.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def get_llm_api_call_cost(
    model: str, prompt_tokens: int, completion_tokens: float
) -> float:
    """
    In older versions of langchain, the cost calculation doesn't work for
    newer models. This is a temporary workaround to get the cost.

    See:
    https://github.com/langchain-ai/langchain/issues/12994

    Costs are in USD, per 1000 tokens.
    """
    model_costs = {
        "gpt-4-1106-preview": {
            "prompt_tokens": 0.01,
            "completion_tokens": 0.03,
        },
        "gpt-4-turbo-2024-04-09": {
            "prompt_tokens": 0.01,
            "completion_tokens": 0.03,
        },
        "gpt-3.5-turbo-0125": {
            "prompt_tokens": 0.0005,
            "completion_tokens": 0.0015,
        },
    }
    if model not in model_costs:
        raise ValueError(f"Unknown model: {model}")

    model_cost = model_costs[model]["prompt_tokens"] * prompt_tokens
    model_cost += model_costs[model]["completion_tokens"] * completion_tokens
    model_cost /= 1000
    return model_cost