Configuration Module

prediction_market_agent_tooling.config

ETHEREUM_ID module-attribute

ETHEREUM_ID = ChainID(1)

GNOSIS_CHAIN_ID module-attribute

GNOSIS_CHAIN_ID = ChainID(100)

ChainID module-attribute

ChainID = NewType('ChainID', int)

PrivateKey module-attribute

PrivateKey = NewType('PrivateKey', SecretStr)

SECRET_TYPES module-attribute

SECRET_TYPES = [
    SecretStr,
    PrivateKey,
    Optional[SecretStr],
    Optional[PrivateKey],
]

APIKeys

Bases: BaseSettings

model_config class-attribute instance-attribute

model_config = SettingsConfigDict(
    env_file=".env",
    env_file_encoding="utf-8",
    extra="ignore",
)

MANIFOLD_API_KEY class-attribute instance-attribute

MANIFOLD_API_KEY: Optional[SecretStr] = None

METACULUS_API_KEY class-attribute instance-attribute

METACULUS_API_KEY: Optional[SecretStr] = None

METACULUS_USER_ID class-attribute instance-attribute

METACULUS_USER_ID: Optional[int] = None

BET_FROM_PRIVATE_KEY class-attribute instance-attribute

BET_FROM_PRIVATE_KEY: Optional[PrivateKey] = None

SAFE_ADDRESS class-attribute instance-attribute

SAFE_ADDRESS: str | None = None

OPENAI_API_KEY class-attribute instance-attribute

OPENAI_API_KEY: Optional[SecretStr] = None

GRAPH_API_KEY class-attribute instance-attribute

GRAPH_API_KEY: Optional[SecretStr] = None

TENDERLY_FORK_RPC class-attribute instance-attribute

TENDERLY_FORK_RPC: Optional[str] = None

GOOGLE_SEARCH_API_KEY class-attribute instance-attribute

GOOGLE_SEARCH_API_KEY: Optional[SecretStr] = None

GOOGLE_SEARCH_ENGINE_ID class-attribute instance-attribute

GOOGLE_SEARCH_ENGINE_ID: Optional[SecretStr] = None

LANGFUSE_SECRET_KEY class-attribute instance-attribute

LANGFUSE_SECRET_KEY: Optional[SecretStr] = None

LANGFUSE_PUBLIC_KEY class-attribute instance-attribute

LANGFUSE_PUBLIC_KEY: Optional[str] = None

LANGFUSE_HOST class-attribute instance-attribute

LANGFUSE_HOST: Optional[str] = None

LANGFUSE_DEPLOYMENT_VERSION class-attribute instance-attribute

LANGFUSE_DEPLOYMENT_VERSION: Optional[str] = None

ENABLE_IPFS_UPLOAD class-attribute instance-attribute

ENABLE_IPFS_UPLOAD: bool = False

PINATA_API_KEY class-attribute instance-attribute

PINATA_API_KEY: Optional[SecretStr] = None

PINATA_API_SECRET class-attribute instance-attribute

PINATA_API_SECRET: Optional[SecretStr] = None

TAVILY_API_KEY class-attribute instance-attribute

TAVILY_API_KEY: Optional[SecretStr] = None

SERPER_API_KEY class-attribute instance-attribute

SERPER_API_KEY: Optional[SecretStr] = None

SQLALCHEMY_DB_URL class-attribute instance-attribute

SQLALCHEMY_DB_URL: Optional[SecretStr] = None

PERPLEXITY_API_KEY class-attribute instance-attribute

PERPLEXITY_API_KEY: Optional[SecretStr] = None

ENABLE_CACHE class-attribute instance-attribute

ENABLE_CACHE: bool = False

CACHE_DIR class-attribute instance-attribute

CACHE_DIR: str = './.cache'

safe_address_checksum property

safe_address_checksum: ChecksumAddress | None

serper_api_key property

serper_api_key: SecretStr

manifold_user_id property

manifold_user_id: str

manifold_api_key property

manifold_api_key: SecretStr

metaculus_api_key property

metaculus_api_key: SecretStr

metaculus_user_id property

metaculus_user_id: int

bet_from_private_key property

bet_from_private_key: PrivateKey

public_key property

public_key: ChecksumAddress

bet_from_address property

bet_from_address: ChecksumAddress

If the SAFE is available, we always route transactions via SAFE. Otherwise we use the EOA.

openai_api_key property

openai_api_key: SecretStr

openai_api_key_secretstr_v1 property

openai_api_key_secretstr_v1: SecretStr

graph_api_key property

graph_api_key: SecretStr

google_search_api_key property

google_search_api_key: SecretStr

google_search_engine_id property

google_search_engine_id: SecretStr

langfuse_secret_key property

langfuse_secret_key: SecretStr

langfuse_public_key property

langfuse_public_key: str

langfuse_host property

langfuse_host: str

default_enable_langfuse property

default_enable_langfuse: bool

enable_ipfs_upload property

enable_ipfs_upload: bool

pinata_api_key property

pinata_api_key: SecretStr

pinata_api_secret property

pinata_api_secret: SecretStr

tavily_api_key property

tavily_api_key: SecretStr

sqlalchemy_db_url property

sqlalchemy_db_url: SecretStr

perplexity_api_key property

perplexity_api_key: SecretStr

copy_without_safe_address

copy_without_safe_address() -> APIKeys

This is handy when you operate in environment with SAFE_ADDRESS, but need to execute transaction using EOA.

Source code in prediction_market_agent_tooling/config.py

def copy_without_safe_address(self) -> "APIKeys":
    """
    This is handy when you operate in environment with SAFE_ADDRESS, but need to execute transaction using EOA.
    """
    data = self.model_copy(deep=True)
    data.SAFE_ADDRESS = None
    return data

get_account

get_account() -> LocalAccount

Source code in prediction_market_agent_tooling/config.py

def get_account(self) -> LocalAccount:
    acc: LocalAccount = Account.from_key(
        self.bet_from_private_key.get_secret_value()
    )
    return acc

model_dump_public

model_dump_public() -> dict[str, t.Any]

Source code in prediction_market_agent_tooling/config.py

def model_dump_public(self) -> dict[str, t.Any]:
    return {
        k: v
        for k, v in self.model_dump().items()
        if self.model_fields[k].annotation not in SECRET_TYPES and v is not None
    }

model_dump_secrets

model_dump_secrets() -> dict[str, t.Any]

Source code in prediction_market_agent_tooling/config.py

def model_dump_secrets(self) -> dict[str, t.Any]:
    return {
        k: v.get_secret_value() if isinstance(v, SecretStr) else v
        for k, v in self.model_dump().items()
        if self.model_fields[k].annotation in SECRET_TYPES and v is not None
    }

check_if_is_safe_owner

check_if_is_safe_owner(
    ethereum_client: EthereumClient,
) -> bool

Source code in prediction_market_agent_tooling/config.py

def check_if_is_safe_owner(self, ethereum_client: EthereumClient) -> bool:
    if not self.safe_address_checksum:
        raise ValueError("Cannot check ownership if safe_address is not defined.")

    s = SafeV141(self.safe_address_checksum, ethereum_client)
    public_key_from_signer = private_key_to_public_key(self.bet_from_private_key)
    return s.retrieve_is_owner(public_key_from_signer)

RPCConfig

Bases: BaseSettings

model_config class-attribute instance-attribute

model_config = SettingsConfigDict(
    env_file=".env",
    env_file_encoding="utf-8",
    extra="ignore",
)

ETHEREUM_RPC_URL class-attribute instance-attribute

ETHEREUM_RPC_URL: URI = Field(
    default=URI("https://rpc.eth.gateway.fm")
)

ETHEREUM_RPC_BEARER class-attribute instance-attribute

ETHEREUM_RPC_BEARER: SecretStr | None = None

GNOSIS_RPC_URL class-attribute instance-attribute

GNOSIS_RPC_URL: URI = Field(
    default=URI("https://rpc.gnosis.gateway.fm")
)

GNOSIS_RPC_BEARER class-attribute instance-attribute

GNOSIS_RPC_BEARER: SecretStr | None = None

CHAIN_ID class-attribute instance-attribute

CHAIN_ID: ChainID = Field(default=GNOSIS_CHAIN_ID)

ethereum_rpc_url property

ethereum_rpc_url: URI

gnosis_rpc_url property

gnosis_rpc_url: URI

chain_id property

chain_id: ChainID

chain_id_to_rpc_url

chain_id_to_rpc_url(chain_id: ChainID) -> URI

Source code in prediction_market_agent_tooling/config.py

def chain_id_to_rpc_url(self, chain_id: ChainID) -> URI:
    return {
        ETHEREUM_ID: self.ethereum_rpc_url,
        GNOSIS_CHAIN_ID: self.gnosis_rpc_url,
    }[chain_id]

chain_id_to_rpc_bearer

chain_id_to_rpc_bearer(
    chain_id: ChainID,
) -> SecretStr | None

Source code in prediction_market_agent_tooling/config.py

def chain_id_to_rpc_bearer(self, chain_id: ChainID) -> SecretStr | None:
    return {
        ETHEREUM_ID: self.ETHEREUM_RPC_BEARER,
        GNOSIS_CHAIN_ID: self.GNOSIS_RPC_BEARER,
    }[chain_id]

get_web3

get_web3() -> Web3

Source code in prediction_market_agent_tooling/config.py

def get_web3(self) -> Web3:
    headers = {
        "Content-Type": "application/json",
        "User-Agent": construct_user_agent(str(type(self))),
    }
    if bearer := self.chain_id_to_rpc_bearer(self.chain_id):
        headers["Authorization"] = f"Bearer {bearer.get_secret_value()}"

    return Web3(
        Web3.HTTPProvider(
            self.chain_id_to_rpc_url(self.chain_id),
            request_kwargs={
                "headers": headers,
            },
        )
    )

CloudCredentials

Bases: BaseSettings

model_config class-attribute instance-attribute

model_config = SettingsConfigDict(
    env_file=".env",
    env_file_encoding="utf-8",
    extra="ignore",
)

GOOGLE_APPLICATION_CREDENTIALS class-attribute instance-attribute

GOOGLE_APPLICATION_CREDENTIALS: Optional[str] = None

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")

secretstr_to_v1_secretstr

secretstr_to_v1_secretstr(s: SecretStr) -> SecretStrV1

secretstr_to_v1_secretstr(s: None) -> None

secretstr_to_v1_secretstr(
    s: SecretStr | None,
) -> SecretStrV1 | None

Source code in prediction_market_agent_tooling/gtypes.py

def secretstr_to_v1_secretstr(s: SecretStr | None) -> SecretStrV1 | None:
    # Another library can be typed with v1, and then we need this ugly conversion.
    return SecretStrV1(s.get_secret_value()) if s is not None else None

get_authenticated_user

get_authenticated_user(api_key: str) -> ManifoldUser

Source code in prediction_market_agent_tooling/markets/manifold/api.py

def get_authenticated_user(api_key: str) -> ManifoldUser:
    url = f"{MANIFOLD_API_BASE_URL}/v0/me"
    headers = {
        "Authorization": f"Key {api_key}",
        "Content-Type": "application/json",
        "Cache-Control": "private, no-store, max-age=0",
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return ManifoldUser.model_validate(response.json())

check_not_none

check_not_none(
    value: Optional[T],
    msg: str = "Value shouldn't be None.",
    exp: Type[ValueError] = ValueError,
) -> T

Utility to remove optionality from a variable.

Useful for cases like this:

keys = pma.utils.get_keys()
pma.omen.omen_buy_outcome_tx(
    from_address=check_not_none(keys.bet_from_address),  # <-- No more Optional[HexAddress], so type checker will be happy.
    ...,
)

Source code in prediction_market_agent_tooling/tools/utils.py

def check_not_none(
    value: Optional[T],
    msg: str = "Value shouldn't be None.",
    exp: Type[ValueError] = ValueError,
) -> T:
    """
    Utility to remove optionality from a variable.

    Useful for cases like this:

    ```
    keys = pma.utils.get_keys()
    pma.omen.omen_buy_outcome_tx(
        from_address=check_not_none(keys.bet_from_address),  # <-- No more Optional[HexAddress], so type checker will be happy.
        ...,
    )
    ```
    """
    if value is None:
        should_not_happen(msg=msg, exp=exp)
    return value

private_key_to_public_key

private_key_to_public_key(
    private_key: SecretStr,
) -> ChecksumAddress

Source code in prediction_market_agent_tooling/tools/web3_utils.py

def private_key_to_public_key(private_key: SecretStr) -> ChecksumAddress:
    account = Account.from_key(private_key.get_secret_value())
    return verify_address(account.address)