Add partial Metaculus support (#299)

prediction_market_agent_tooling/config.py

@@ -24,6 +24,8 @@ class APIKeys(BaseSettings):
     )
 
     MANIFOLD_API_KEY: t.Optional[SecretStr] = None
+    METACULUS_API_KEY: t.Optional[SecretStr] = None
+    METACULUS_USER_ID: t.Optional[int] = None
     BET_FROM_PRIVATE_KEY: t.Optional[PrivateKey] = None
     SAFE_ADDRESS: t.Optional[ChecksumAddress] = None
     OPENAI_API_KEY: t.Optional[SecretStr] = None
@@ -51,6 +53,18 @@ def manifold_api_key(self) -> SecretStr:
             self.MANIFOLD_API_KEY, "MANIFOLD_API_KEY missing in the environment."
         )
 
+    @property
+    def metaculus_api_key(self) -> SecretStr:
+        return check_not_none(
+            self.METACULUS_API_KEY, "METACULUS_API_KEY missing in the environment."
+        )
+
+    @property
+    def metaculus_user_id(self) -> int:
+        return check_not_none(
+            self.METACULUS_USER_ID, "METACULUS_USER_ID missing in the environment."
+        )
+
     @property
     def bet_from_private_key(self) -> PrivateKey:
         return check_not_none(

prediction_market_agent_tooling/markets/markets.py

@@ -16,6 +16,9 @@
 from prediction_market_agent_tooling.markets.manifold.manifold import (
     ManifoldAgentMarket,
 )
+from prediction_market_agent_tooling.markets.metaculus.metaculus import (
+    MetaculusAgentMarket,
+)
 from prediction_market_agent_tooling.markets.omen.omen import OmenAgentMarket
 from prediction_market_agent_tooling.markets.omen.omen_subgraph_handler import (
     OmenSubgraphHandler,
@@ -30,6 +33,7 @@ class MarketType(str, Enum):
     MANIFOLD = "manifold"
     OMEN = "omen"
     POLYMARKET = "polymarket"
+    METACULUS = "metaculus"
 
     @property
     def market_class(self) -> type[AgentMarket]:
@@ -42,6 +46,7 @@ def market_class(self) -> type[AgentMarket]:
     MarketType.MANIFOLD: ManifoldAgentMarket,
     MarketType.OMEN: OmenAgentMarket,
     MarketType.POLYMARKET: PolymarketAgentMarket,
+    MarketType.METACULUS: MetaculusAgentMarket,
 }
 
 

prediction_market_agent_tooling/markets/metaculus/api.py

@@ -0,0 +1,97 @@
+from datetime import datetime
+from typing import Union
+
+import requests
+
+from prediction_market_agent_tooling.config import APIKeys
+from prediction_market_agent_tooling.gtypes import Probability
+from prediction_market_agent_tooling.markets.metaculus.data_models import (
+    MetaculusQuestion,
+)
+from prediction_market_agent_tooling.tools.utils import (
+    response_list_to_model,
+    response_to_model,
+)
+
+METACULUS_API_BASE_URL = "https://www.metaculus.com/api2"
+
+
+def get_auth_headers() -> dict[str, str]:
+    return {"Authorization": f"Token {APIKeys().metaculus_api_key.get_secret_value()}"}
+
+
+def post_question_comment(question_id: str, comment_text: str) -> None:
+    """
+    Post a comment on the question page as the bot user.
+    """
+
+    response = requests.post(
+        f"{METACULUS_API_BASE_URL}/comments/",
+        json={
+            "comment_text": comment_text,
+            "submit_type": "N",
+            "include_latest_prediction": True,
+            "question": question_id,
+        },
+        headers=get_auth_headers(),
+    )
+    response.raise_for_status()
+
+
+def make_prediction(question_id: str, p_yes: Probability) -> None:
+    """
+    Make a prediction for a question.
+    """
+    url = f"{METACULUS_API_BASE_URL}/questions/{question_id}/predict/"
+    response = requests.post(
+        url,
+        json={"prediction": p_yes},
+        headers=get_auth_headers(),
+    )
+    response.raise_for_status()
+
+
+def get_question(question_id: str) -> MetaculusQuestion:
+    """
+    Get all details about a specific question.
+    """
+    url = f"{METACULUS_API_BASE_URL}/questions/{question_id}/"
+    return response_to_model(
+        response=requests.get(url, headers=get_auth_headers()),
+        model=MetaculusQuestion,
+    )
+
+
+def get_questions(
+    limit: int,
+    order_by: str | None = None,
+    offset: int = 0,
+    tournament_id: int | None = None,
+    created_after: datetime | None = None,
+    status: str | None = None,
+) -> list[MetaculusQuestion]:
+    """
+    List detailed metaculus questions (i.e. markets)
+    """
+    url_params: dict[str, Union[int, str]] = {
+        "limit": limit,
+        "offset": offset,
+        "has_group": "false",
+        "forecast_type": "binary",
+        "type": "forecast",
+        "include_description": "true",
+    }
+    if order_by:
+        url_params["order_by"] = order_by
+    if tournament_id:
+        url_params["project"] = tournament_id
+    if created_after:
+        url_params["created_time__gt"] = created_after.isoformat()
+    if status:
+        url_params["status"] = status
+
+    url = f"{METACULUS_API_BASE_URL}/questions/"
+    return response_list_to_model(
+        response=requests.get(url, headers=get_auth_headers(), params=url_params),
+        model=MetaculusQuestion,
+    )

prediction_market_agent_tooling/markets/metaculus/data_models.py

@@ -0,0 +1,90 @@
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class QuestionType(str, Enum):
+    forecast = "forecast"
+    notebook = "notebook"
+    discussion = "discussion"
+    claim = "claim"
+    group = "group"
+    conditional_group = "conditional_group"
+    multiple_choice = "multiple_choice"
+
+
+class CommunityPrediction(BaseModel):
+    y: list[float]
+    q1: float | None = None
+    q2: float | None = None
+    q3: float | None = None
+
+    @property
+    def p_yes(self) -> float:
+        """
+        q2 corresponds to the median, or 'second quartile' of the distribution.
+
+        If no value is provided (i.e. the question is new and has not been
+        answered yet), we default to 0.5.
+        """
+        return self.q2 if self.q2 is not None else 0.5
+
+
+class Prediction(BaseModel):
+    t: datetime
+    x: float
+
+
+class UserPredictions(BaseModel):
+    id: int
+    predictions: list[Prediction]
+    points_won: float | None = None
+    user: int
+    username: str
+    question: int
+
+
+class CommunityPredictionStats(BaseModel):
+    full: CommunityPrediction
+    unweighted: CommunityPrediction
+
+
+class MetaculusQuestion(BaseModel):
+    """
+    https://www.metaculus.com/api2/schema/redoc/#tag/questions/operation/questions_retrieve
+    """
+
+    active_state: Any
+    url: str
+    page_url: str
+    id: int
+    author: int
+    author_name: str
+    author_id: int
+    title: str
+    title_short: str
+    group_label: str | None = None
+    resolution: int | None
+    resolved_option: int | None
+    created_time: datetime
+    publish_time: datetime | None = None
+    close_time: datetime | None = None
+    effected_close_time: datetime | None
+    resolve_time: datetime | None = None
+    possibilities: dict[Any, Any] | None = None
+    scoring: dict[Any, Any] = {}
+    type: QuestionType | None = None
+    user_perms: Any
+    weekly_movement: float | None
+    weekly_movement_direction: int | None = None
+    cp_reveal_time: datetime | None = None
+    edited_time: datetime
+    last_activity_time: datetime
+    activity: float
+    comment_count: int
+    votes: int
+    community_prediction: CommunityPredictionStats
+    my_predictions: UserPredictions | None = None
+    # TODO add the rest of the fields https://github.com/gnosis/prediction-market-agent-tooling/issues/301

prediction_market_agent_tooling/markets/metaculus/metaculus.py

@@ -0,0 +1,102 @@
+import typing as t
+from datetime import datetime
+
+from prediction_market_agent_tooling.gtypes import Probability
+from prediction_market_agent_tooling.markets.agent_market import (
+    AgentMarket,
+    FilterBy,
+    SortBy,
+)
+from prediction_market_agent_tooling.markets.metaculus.api import (
+    METACULUS_API_BASE_URL,
+    get_questions,
+    make_prediction,
+    post_question_comment,
+)
+from prediction_market_agent_tooling.markets.metaculus.data_models import (
+    MetaculusQuestion,
+)
+
+
+class MetaculusAgentMarket(AgentMarket):
+    """
+    Metaculus' market class that can be used by agents to make predictions.
+    """
+
+    have_predicted: bool
+    base_url: t.ClassVar[str] = METACULUS_API_BASE_URL
+
+    @staticmethod
+    def from_data_model(model: MetaculusQuestion) -> "MetaculusAgentMarket":
+        return MetaculusAgentMarket(
+            id=str(model.id),
+            question=model.title,
+            outcomes=[],
+            resolution=None,
+            current_p_yes=Probability(model.community_prediction.full.p_yes),
+            created_time=model.created_time,
+            close_time=model.close_time,
+            url=model.url,
+            volume=None,
+            have_predicted=model.my_predictions is not None
+            and len(model.my_predictions.predictions) > 0,
+        )
+
+    @staticmethod
+    def get_binary_markets(
+        limit: int,
+        sort_by: SortBy = SortBy.NONE,
+        filter_by: FilterBy = FilterBy.OPEN,
+        created_after: t.Optional[datetime] = None,
+        excluded_questions: set[str] | None = None,
+        tournament_id: int | None = None,
+    ) -> t.Sequence["MetaculusAgentMarket"]:
+        order_by: str | None
+        if sort_by == SortBy.NONE:
+            order_by = None
+        elif sort_by == SortBy.CLOSING_SOONEST:
+            order_by = "-close_time"
+        elif sort_by == SortBy.NEWEST:
+            order_by = "-created_time"
+        else:
+            raise ValueError(f"Unknown sort_by: {sort_by}")
+
+        status: str | None
+        if filter_by == FilterBy.OPEN:
+            status = "open"
+        elif filter_by == FilterBy.RESOLVED:
+            status = "resolved"
+        elif filter_by == FilterBy.NONE:
+            status = None
+        else:
+            raise ValueError(f"Unknown filter_by: {filter_by}")
+
+        if excluded_questions:
+            raise NotImplementedError(
+                "Excluded questions are not suppoerted for Metaculus markets yet."
+            )
+
+        offset = 0
+        question_page_size = 500
+        all_questions = []
+        while True:
+            questions = get_questions(
+                limit=question_page_size,
+                offset=offset,
+                order_by=order_by,
+                created_after=created_after,
+                status=status,
+                tournament_id=tournament_id,
+            )
+            if not questions:
+                break
+            all_questions.extend(questions)
+            offset += question_page_size
+
+            if len(all_questions) >= limit:
+                break
+        return [MetaculusAgentMarket.from_data_model(q) for q in all_questions]
+
+    def submit_prediction(self, p_yes: Probability, reasoning: str) -> None:
+        make_prediction(self.id, p_yes)
+        post_question_comment(self.id, reasoning)

prediction_market_agent_tooling/monitor/markets/metaculus.py

@@ -0,0 +1,43 @@
+import typing as t
+
+from google.cloud.functions_v2.types.functions import Function
+
+from prediction_market_agent_tooling.config import APIKeys
+from prediction_market_agent_tooling.deploy.constants import MARKET_TYPE_KEY
+from prediction_market_agent_tooling.gtypes import DatetimeWithTimezone
+from prediction_market_agent_tooling.markets.data_models import ResolvedBet
+from prediction_market_agent_tooling.markets.markets import MarketType
+from prediction_market_agent_tooling.monitor.monitor import DeployedAgent
+
+
+class DeployedMetaculusAgent(DeployedAgent):
+    user: int
+
+    @property
+    def public_id(self) -> str:
+        return str(self.user)
+
+    def get_resolved_bets(self) -> list[ResolvedBet]:
+        raise NotImplementedError("TODO: Implement to allow betting on Metaculus.")
+
+    @staticmethod
+    def from_api_keys(
+        name: str,
+        start_time: DatetimeWithTimezone,
+        api_keys: APIKeys,
+    ) -> "DeployedMetaculusAgent":
+        return DeployedMetaculusAgent(
+            name=name,
+            start_time=start_time,
+            user=api_keys.metaculus_user_id,
+        )
+
+    @classmethod
+    def from_all_gcp_functions(
+        cls: t.Type["DeployedMetaculusAgent"],
+        filter_: t.Callable[[Function], bool] = lambda function: function.labels[
+            MARKET_TYPE_KEY
+        ]
+        == MarketType.METACULUS.value,
+    ) -> t.Sequence["DeployedMetaculusAgent"]:
+        return super().from_all_gcp_functions(filter_=filter_)