Skip to content

Tools Module

prediction_market_agent_tooling.tools

_generic_value

InputValueType module-attribute 

InputValueType = TypeVar(
    "InputValueType",
    bound=Union[str, int, float, Wei, Decimal],
)

InternalValueType module-attribute 

InternalValueType = TypeVar(
    "InternalValueType", bound=Union[int, float, Wei]
)

balances

Balances

Bases: BaseModel

xdai instance-attribute 
xdai: xDai
wxdai instance-attribute 
wxdai: CollateralToken
sdai instance-attribute 
sdai: CollateralToken
total property 
total: CollateralToken

get_balances 

get_balances(
    address: ChecksumAddress, web3: Web3 | None = None
) -> Balances
Source code in prediction_market_agent_tooling/tools/balances.py 
27
28
29
30
31
32
33
34
35
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def get_balances(address: ChecksumAddress, web3: Web3 | None = None) -> Balances:
    if not web3:
        web3 = WrappedxDaiContract().get_web3()
    xdai_balance = xDaiWei(web3.eth.get_balance(address))
    xdai = xdai_balance.as_xdai
    wxdai = WrappedxDaiContract().balanceOf(address, web3=web3).as_token
    sdai = sDaiContract().balanceOf(address, web3=web3).as_token
    return Balances(xdai=xdai, wxdai=wxdai, sdai=sdai)

betting_strategies

kelly_criterion

check_is_valid_probability 
check_is_valid_probability(probability: float) -> None
Source code in prediction_market_agent_tooling/tools/betting_strategies/kelly_criterion.py 
6
7
8
def check_is_valid_probability(probability: float) -> None:
    if not 0 <= probability <= 1:
        raise ValueError("Probability must be between 0 and 1")
get_kelly_bet_simplified 
get_kelly_bet_simplified(
    max_bet: CollateralToken,
    market_p_yes: float,
    estimated_p_yes: float,
    confidence: float,
) -> SimpleBet

Calculate the optimal bet amount using the Kelly Criterion for a binary outcome market.

From https://en.wikipedia.org/wiki/Kelly_criterion:

f* = p - q / b

where: - f* is the fraction of the current bankroll to wager - p is the probability of a win - q = 1-p is the probability of a loss - b is the proportion of the bet gained with a win

Note: this calculation does not factor in that the bet changes the market odds. This means the calculation is only accurate if the bet size is small compared to the market volume. See discussion here for more detail: https://github.com/gnosis/prediction-market-agent-tooling/pull/330#discussion_r1698269328

Source code in prediction_market_agent_tooling/tools/betting_strategies/kelly_criterion.py 
11
12
13
14
15
16
17
18
19
20
21
22
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
def get_kelly_bet_simplified(
    max_bet: CollateralToken,
    market_p_yes: float,
    estimated_p_yes: float,
    confidence: float,
) -> SimpleBet:
    """
    Calculate the optimal bet amount using the Kelly Criterion for a binary outcome market.

    From https://en.wikipedia.org/wiki/Kelly_criterion:

    f* = p - q / b

    where:
    - f* is the fraction of the current bankroll to wager
    - p is the probability of a win
    - q = 1-p is the probability of a loss
    - b is the proportion of the bet gained with a win

    Note: this calculation does not factor in that the bet changes the market
    odds. This means the calculation is only accurate if the bet size is small
    compared to the market volume. See discussion here for more detail:
    https://github.com/gnosis/prediction-market-agent-tooling/pull/330#discussion_r1698269328
    """
    check_is_valid_probability(market_p_yes)
    check_is_valid_probability(estimated_p_yes)
    check_is_valid_probability(confidence)

    if estimated_p_yes > market_p_yes:
        bet_direction = True
        market_prob = market_p_yes
    else:
        bet_direction = False
        market_prob = 1 - market_p_yes

    # Handle the case where market_prob is 0
    if market_prob == 0:
        market_prob = 1e-10

    edge = abs(estimated_p_yes - market_p_yes) * confidence
    odds = (1 / market_prob) - 1
    kelly_fraction = edge / odds

    # Ensure bet size is non-negative does not exceed the wallet balance
    bet_size = CollateralToken(min(kelly_fraction * max_bet.value, max_bet.value))

    return SimpleBet(direction=bet_direction, size=bet_size)
get_kelly_bet_full 
get_kelly_bet_full(
    yes_outcome_pool_size: OutcomeToken,
    no_outcome_pool_size: OutcomeToken,
    estimated_p_yes: float,
    confidence: float,
    max_bet: CollateralToken,
    fees: MarketFees,
) -> SimpleBet

Calculate the optimal bet amount using the Kelly Criterion for a binary outcome market.

'Full' as in it accounts for how the bet changes the market odds.

Taken from https://github.com/valory-xyz/trader/blob/main/strategies/kelly_criterion/kelly_criterion.py

with derivation in PR description: https://github.com/valory-xyz/trader/pull/119

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Source code in prediction_market_agent_tooling/tools/betting_strategies/kelly_criterion.py 
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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
def get_kelly_bet_full(
    yes_outcome_pool_size: OutcomeToken,
    no_outcome_pool_size: OutcomeToken,
    estimated_p_yes: float,
    confidence: float,
    max_bet: CollateralToken,
    fees: MarketFees,
) -> SimpleBet:
    """
    Calculate the optimal bet amount using the Kelly Criterion for a binary outcome market.

    'Full' as in it accounts for how the bet changes the market odds.

    Taken from https://github.com/valory-xyz/trader/blob/main/strategies/kelly_criterion/kelly_criterion.py

    with derivation in PR description: https://github.com/valory-xyz/trader/pull/119

    ```
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
    ```
    """
    fee = fees.bet_proportion
    if fees.absolute > 0:
        raise RuntimeError(
            f"Kelly works only with bet-proportional fees, but the fees are {fees=}."
        )

    check_is_valid_probability(estimated_p_yes)
    check_is_valid_probability(confidence)

    if max_bet == 0:
        return SimpleBet(size=CollateralToken(0), direction=True)

    x = yes_outcome_pool_size.value
    y = no_outcome_pool_size.value
    p = estimated_p_yes
    c = confidence
    b = max_bet.value
    f = 1 - fee

    if x == y:
        # Add a delta to prevent division by zero
        y += 1e-10

    numerator = (
        -4 * x**2 * y
        + b * y**2 * p * c * f
        + 2 * b * x * y * p * c * f
        + b * x**2 * p * c * f
        - 2 * b * y**2 * f
        - 2 * b * x * y * f
        + (
            (
                4 * x**2 * y
                - b * y**2 * p * c * f
                - 2 * b * x * y * p * c * f
                - b * x**2 * p * c * f
                + 2 * b * y**2 * f
                + 2 * b * x * y * f
            )
            ** 2
            - (
                4
                * (x**2 * f - y**2 * f)
                * (
                    -4 * b * x * y**2 * p * c
                    - 4 * b * x**2 * y * p * c
                    + 4 * b * x * y**2
                )
            )
        )
        ** (1 / 2)
    )
    denominator = 2 * (x**2 * f - y**2 * f)
    kelly_bet_amount = numerator / denominator

    # Clip the bet size to max_bet to account for rounding errors.
    return SimpleBet(
        direction=kelly_bet_amount > 0,
        size=CollateralToken(min(max_bet.value, abs(kelly_bet_amount))),
    )

stretch_bet_between

stretch_bet_between 
stretch_bet_between(
    probability: Probability, min_bet: float, max_bet: float
) -> float

Normalise the outcome probability into a bet amount between the minimum and maximum bet.

Source code in prediction_market_agent_tooling/tools/betting_strategies/stretch_bet_between.py 
 4
 5
 6
 7
 8
 9
10
11
12
def stretch_bet_between(
    probability: Probability, min_bet: float, max_bet: float
) -> float:
    """
    Normalise the outcome probability into a bet amount between the minimum and maximum bet.
    """
    if min_bet > max_bet:
        raise ValueError("Minimum bet cannot be greater than maximum bet.")
    return min_bet + (max_bet - min_bet) * probability

utils

SimpleBet

Bases: BaseModel

direction instance-attribute 
direction: bool
size instance-attribute 
size: CollateralToken

caches

db_cache

DB_CACHE_LOG_PREFIX module-attribute 
DB_CACHE_LOG_PREFIX = '[db-cache]'
FunctionT module-attribute 
FunctionT = TypeVar('FunctionT', bound=Callable[..., Any])
FunctionCache

Bases: SQLModel

id class-attribute instance-attribute 
id: int | None = Field(default=None, primary_key=True)
function_name class-attribute instance-attribute 
function_name: str = Field(index=True)
full_function_name class-attribute instance-attribute 
full_function_name: str = Field(index=True)
args class-attribute instance-attribute 
args: Any = Field(sa_column=Column(JSONB, nullable=False))
args_hash class-attribute instance-attribute 
args_hash: str = Field(index=True)
result class-attribute instance-attribute 
result: Any = Field(sa_column=Column(JSONB, nullable=False))
created_at class-attribute instance-attribute 
created_at: DatetimeUTC = Field(
    default_factory=utcnow, index=True
)
db_cache 
db_cache(
    func: None = None,
    *,
    max_age: timedelta | None = None,
    cache_none: bool = True,
    api_keys: APIKeys | None = None,
    ignore_args: Sequence[str] | None = None,
    ignore_arg_types: Sequence[type] | None = None,
    log_error_on_unsavable_data: bool = True,
) -> Callable[[FunctionT], FunctionT]

db_cache(
    func: FunctionT,
    *,
    max_age: timedelta | None = None,
    cache_none: bool = True,
    api_keys: APIKeys | None = None,
    ignore_args: Sequence[str] | None = None,
    ignore_arg_types: Sequence[type] | None = None,
    log_error_on_unsavable_data: bool = True,
) -> FunctionT

db_cache(
    func: FunctionT | None = None,
    *,
    max_age: timedelta | None = None,
    cache_none: bool = True,
    api_keys: APIKeys | None = None,
    ignore_args: Sequence[str] | None = None,
    ignore_arg_types: Sequence[type] | None = None,
    log_error_on_unsavable_data: bool = True,
) -> FunctionT | Callable[[FunctionT], FunctionT]
Source code in prediction_market_agent_tooling/tools/caches/db_cache.py 
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def db_cache(
    func: FunctionT | None = None,
    *,
    max_age: timedelta | None = None,
    cache_none: bool = True,
    api_keys: APIKeys | None = None,
    ignore_args: Sequence[str] | None = None,
    ignore_arg_types: Sequence[type] | None = None,
    log_error_on_unsavable_data: bool = True,
) -> FunctionT | Callable[[FunctionT], FunctionT]:
    if func is None:
        # Ugly Pythonic way to support this decorator as `@postgres_cache` but also `@postgres_cache(max_age=timedelta(days=3))`
        def decorator(func: FunctionT) -> FunctionT:
            return db_cache(
                func,
                max_age=max_age,
                cache_none=cache_none,
                api_keys=api_keys,
                ignore_args=ignore_args,
                ignore_arg_types=ignore_arg_types,
                log_error_on_unsavable_data=log_error_on_unsavable_data,
            )

        return decorator

    api_keys = api_keys if api_keys is not None else APIKeys()

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        # If caching is disabled, just call the function and return it
        if not api_keys.ENABLE_CACHE:
            return func(*args, **kwargs)

        DBManager(api_keys.sqlalchemy_db_url.get_secret_value()).create_tables(
            [FunctionCache]
        )

        # Convert *args and **kwargs to a single dictionary, where we have names for arguments passed as args as well.
        signature = inspect.signature(func)
        bound_arguments = signature.bind(*args, **kwargs)
        bound_arguments.apply_defaults()

        # Convert any argument that is Pydantic model into classic dictionary, otherwise it won't be json-serializable.
        args_dict: dict[str, Any] = bound_arguments.arguments

        # Remove `self` or `cls` if present (in case of class' methods)
        if "self" in args_dict:
            del args_dict["self"]
        if "cls" in args_dict:
            del args_dict["cls"]

        # Remove ignored arguments
        if ignore_args:
            for arg in ignore_args:
                if arg in args_dict:
                    del args_dict[arg]

        # Remove arguments of ignored types
        if ignore_arg_types:
            args_dict = {
                k: v
                for k, v in args_dict.items()
                if not isinstance(v, tuple(ignore_arg_types))
            }

        # Compute a hash of the function arguments used for lookup of cached results
        arg_string = json.dumps(args_dict, sort_keys=True, default=str)
        args_hash = hashlib.md5(arg_string.encode()).hexdigest()

        # Get the full function name as concat of module and qualname, to not accidentally clash
        full_function_name = func.__module__ + "." + func.__qualname__
        # But also get the standard function name to easily search for it in database
        function_name = func.__name__

        # Determine if the function returns or contains Pydantic BaseModel(s)
        return_type = func.__annotations__.get("return", None)
        is_pydantic_model = return_type is not None and contains_pydantic_model(
            return_type
        )

        with DBManager(
            api_keys.sqlalchemy_db_url.get_secret_value()
        ).get_session() as session:
            # Try to get cached result
            statement = (
                select(FunctionCache)
                .where(
                    FunctionCache.function_name == function_name,
                    FunctionCache.full_function_name == full_function_name,
                    FunctionCache.args_hash == args_hash,
                )
                .order_by(desc(FunctionCache.created_at))
            )
            if max_age is not None:
                cutoff_time = utcnow() - max_age
                statement = statement.where(FunctionCache.created_at >= cutoff_time)
            cached_result = session.exec(statement).first()

        if cached_result:
            logger.info(
                # Keep the special [case-hit] identifier so we can easily track it in GCP.
                f"{DB_CACHE_LOG_PREFIX} [cache-hit] Cache hit for {full_function_name} with args {args_dict} and output {cached_result.result}"
            )
            if is_pydantic_model:
                # If the output contains any Pydantic models, we need to initialise them.
                try:
                    return convert_cached_output_to_pydantic(
                        return_type, cached_result.result
                    )
                except ValueError as e:
                    # In case of backward-incompatible pydantic model, just treat it as cache miss, to not error out.
                    logger.warning(
                        f"{DB_CACHE_LOG_PREFIX} [cache-miss] Can not validate {cached_result=} into {return_type=} because {e=}, treating as cache miss."
                    )
                    cached_result = None
            else:
                return cached_result.result

        # On cache miss, compute the result
        computed_result = func(*args, **kwargs)
        # Keep the special [case-miss] identifier so we can easily track it in GCP.
        logger.info(
            f"{DB_CACHE_LOG_PREFIX} [cache-miss] Cache miss for {full_function_name} with args {args_dict}, computed the output {computed_result}"
        )

        # If postgres access was specified, save it.
        if cache_none or computed_result is not None:
            cache_entry = FunctionCache(
                function_name=function_name,
                full_function_name=full_function_name,
                args_hash=args_hash,
                args=args_dict,
                result=computed_result,
                created_at=utcnow(),
            )
            # Do not raise an exception if saving to the database fails, just log it and let the agent continue the work.
            try:
                with DBManager(
                    api_keys.sqlalchemy_db_url.get_secret_value()
                ).get_session() as session:
                    logger.info(
                        f"{DB_CACHE_LOG_PREFIX} [cache-info] Saving {cache_entry} into database."
                    )
                    session.add(cache_entry)
                    session.commit()
            except (DataError, psycopg2.errors.UntranslatableCharacter) as e:
                (logger.error if log_error_on_unsavable_data else logger.warning)(
                    f"{DB_CACHE_LOG_PREFIX} [cache-error] Failed to save {cache_entry} into database, ignoring, because: {e}"
                )
            except Exception:
                logger.exception(
                    f"{DB_CACHE_LOG_PREFIX} [cache-error] Failed to save {cache_entry} into database, ignoring."
                )

        return computed_result

    return cast(FunctionT, wrapper)
contains_pydantic_model 
contains_pydantic_model(return_type: Any) -> bool

Check if the return type contains anything that's a Pydantic model (including nested structures, like list[BaseModel], dict[str, list[BaseModel]], etc.)

Source code in prediction_market_agent_tooling/tools/caches/db_cache.py 
236
237
238
239
240
241
242
243
244
245
246
247
def contains_pydantic_model(return_type: Any) -> bool:
    """
    Check if the return type contains anything that's a Pydantic model (including nested structures, like `list[BaseModel]`, `dict[str, list[BaseModel]]`, etc.)
    """
    if return_type is None:
        return False
    origin = get_origin(return_type)
    if origin is not None:
        return any(contains_pydantic_model(arg) for arg in get_args(return_type))
    if inspect.isclass(return_type):
        return issubclass(return_type, BaseModel)
    return False
convert_cached_output_to_pydantic 
convert_cached_output_to_pydantic(
    return_type: Any, data: Any
) -> Any

Used to initialize Pydantic models from anything cached that was originally a Pydantic model in the output. Including models in nested structures.

Source code in prediction_market_agent_tooling/tools/caches/db_cache.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def convert_cached_output_to_pydantic(return_type: Any, data: Any) -> Any:
    """
    Used to initialize Pydantic models from anything cached that was originally a Pydantic model in the output. Including models in nested structures.
    """
    # Get the origin and arguments of the model type
    origin = get_origin(return_type)
    args = get_args(return_type)

    # Check if the data is a dictionary
    if isinstance(data, dict):
        # If the model has no origin, check if it is a subclass of BaseModel
        if origin is None:
            if inspect.isclass(return_type) and issubclass(return_type, BaseModel):
                # Convert the dictionary to a Pydantic model
                return return_type(
                    **{
                        k: convert_cached_output_to_pydantic(
                            getattr(return_type, k, None), v
                        )
                        for k, v in data.items()
                    }
                )
            else:
                # If not a Pydantic model, return the data as is
                return data
        # If the origin is a dictionary, convert keys and values
        elif origin is dict:
            key_type, value_type = args
            return {
                convert_cached_output_to_pydantic(
                    key_type, k
                ): convert_cached_output_to_pydantic(value_type, v)
                for k, v in data.items()
            }
        # If the origin is a union and one of the unions is basemodel, convert it to it.
        elif (
            origin is UnionType
            and (
                base_model_from_args := next(
                    (x for x in args if issubclass(x, BaseModel)), None
                )
            )
            is not None
        ):
            return base_model_from_args.model_validate(data)
        else:
            # If the origin is not a dictionary, return the data as is
            return data
    # Check if the data is a list
    elif isinstance(data, (list, tuple)):
        # If the origin is a list or tuple, convert each item
        if origin in {list, tuple}:
            item_type = args[0]
            converted_items = [
                convert_cached_output_to_pydantic(item_type, item) for item in data
            ]
            return type(data)(converted_items)
        else:
            # If the origin is not a list or tuple, return the data as is
            return data
    else:
        # If the data is neither a dictionary nor a list, return it as is
        return data

inmemory_cache

MEMORY module-attribute 
MEMORY = Memory(CACHE_DIR, verbose=0)
T module-attribute 
T = TypeVar('T', bound=Callable[..., Any])
persistent_inmemory_cache 
persistent_inmemory_cache(
    func: None = None, *, in_memory_cache: bool = True
) -> Callable[[T], T]

persistent_inmemory_cache(
    func: T, *, in_memory_cache: bool = True
) -> T

persistent_inmemory_cache(
    func: T | None = None, *, in_memory_cache: bool = True
) -> T | Callable[[T], T]

Wraps a function with both file cache (for persistent cache) and optional in-memory cache (for speed). Can be used as @persistent_inmemory_cache or @persistent_inmemory_cache(in_memory_cache=False)

Source code in prediction_market_agent_tooling/tools/caches/inmemory_cache.py 
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
def persistent_inmemory_cache(
    func: T | None = None,
    *,
    in_memory_cache: bool = True,
) -> T | Callable[[T], T]:
    """
    Wraps a function with both file cache (for persistent cache) and optional in-memory cache (for speed).
    Can be used as @persistent_inmemory_cache or @persistent_inmemory_cache(in_memory_cache=False)
    """
    if func is None:
        # Ugly Pythonic way to support this decorator as `@persistent_inmemory_cache` but also `@persistent_inmemory_cache(in_memory_cache=False)`
        def decorator(func: T) -> T:
            return persistent_inmemory_cache(
                func,
                in_memory_cache=in_memory_cache,
            )

        return decorator
    else:
        # The decorator is called without arguments.
        if not APIKeys().ENABLE_CACHE:
            return func
        cached_func = MEMORY.cache(func)
        if in_memory_cache:
            cached_func = cache(cached_func)
        return cast(T, cached_func)

serializers

json_serializer 
json_serializer(x: Any) -> str
Source code in prediction_market_agent_tooling/tools/caches/serializers.py 
10
11
def json_serializer(x: t.Any) -> str:
    return json.dumps(x, default=json_serializer_default_fn)
json_serializer_default_fn 
json_serializer_default_fn(
    y: DatetimeUTC | timedelta | date | BaseModel,
) -> str | dict[str, t.Any]

Used to serialize objects that don't support it by default into a specific string that can be deserialized out later. If this function returns a dictionary, it will be called recursively. If you add something here, also add it to replace_custom_stringified_objects below.

Source code in prediction_market_agent_tooling/tools/caches/serializers.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def json_serializer_default_fn(
    y: DatetimeUTC | timedelta | date | BaseModel,
) -> str | dict[str, t.Any]:
    """
    Used to serialize objects that don't support it by default into a specific string that can be deserialized out later.
    If this function returns a dictionary, it will be called recursively.
    If you add something here, also add it to `replace_custom_stringified_objects` below.
    """
    if isinstance(y, DatetimeUTC):
        return f"DatetimeUTC::{y.isoformat()}"
    elif isinstance(y, timedelta):
        return f"timedelta::{y.total_seconds()}"
    elif isinstance(y, date):
        return f"date::{y.isoformat()}"
    elif isinstance(y, BaseModel):
        return y.model_dump()
    raise TypeError(
        f"Unsupported type for the default json serialize function, value is {y}."
    )
json_deserializer 
json_deserializer(s: str) -> t.Any
Source code in prediction_market_agent_tooling/tools/caches/serializers.py 
35
36
37
def json_deserializer(s: str) -> t.Any:
    data = json.loads(s)
    return replace_custom_stringified_objects(data)
replace_custom_stringified_objects 
replace_custom_stringified_objects(obj: Any) -> t.Any

Used to deserialize objects from json_serializer_default_fn into their proper form.

Source code in prediction_market_agent_tooling/tools/caches/serializers.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def replace_custom_stringified_objects(obj: t.Any) -> t.Any:
    """
    Used to deserialize objects from `json_serializer_default_fn` into their proper form.
    """
    if isinstance(obj, str):
        if obj.startswith("DatetimeUTC::"):
            iso_str = obj[len("DatetimeUTC::") :]
            return DatetimeUTC.to_datetime_utc(iso_str)
        elif obj.startswith("timedelta::"):
            total_seconds_str = obj[len("timedelta::") :]
            return timedelta(seconds=float(total_seconds_str))
        elif obj.startswith("date::"):
            iso_str = obj[len("date::") :]
            return date.fromisoformat(iso_str)
        else:
            return obj
    elif isinstance(obj, dict):
        return {k: replace_custom_stringified_objects(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [replace_custom_stringified_objects(item) for item in obj]
    else:
        return obj

contract

ContractBaseClass

Bases: BaseModel

Base class holding the basic requirements and tools used for every contract.

CHAIN_ID class-attribute 
CHAIN_ID: ChainID
abi instance-attribute 
abi: ABI
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

ContractProxyBaseClass

Bases: ContractBaseClass

Contract base class for proxy contracts.

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/proxy.abi.json",
    )
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
implementation 
implementation(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
184
185
186
def implementation(self, web3: Web3 | None = None) -> ChecksumAddress:
    address = self.call("implementation", web3=web3)
    return Web3.to_checksum_address(address)
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

ContractERC20BaseClass

Bases: ContractBaseClass

Contract base class extended by ERC-20 standard methods.

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/erc20.abi.json",
    )
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
200
201
202
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
symbol_cached 
symbol_cached(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
204
205
206
207
208
209
210
def symbol_cached(self, web3: Web3 | None = None) -> str:
    web3 = web3 or self.get_web3()
    cache_key = create_contract_method_cache_key(self.symbol, web3)
    if cache_key not in self._cache:
        self._cache[cache_key] = self.symbol(web3=web3)
    value: str = self._cache[cache_key]
    return value
allowance 
allowance(
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
212
213
214
215
216
217
218
219
220
221
def allowance(
    self,
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei:
    allowance_for_user: int = self.call(
        "allowance", function_params=[owner, for_address], web3=web3
    )
    return Wei(allowance_for_user)
approve 
approve(
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def approve(
    self,
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="approve",
        function_params=[
            for_address,
            amount_wei,
        ],
        tx_params=tx_params,
        web3=web3,
    )
transferFrom 
transferFrom(
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def transferFrom(
    self,
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="transferFrom",
        function_params=[sender, recipient, amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
259
260
261
def balanceOf(self, for_address: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [for_address], web3=web3))
    return balance
balance_of_in_tokens 
balance_of_in_tokens(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken
Source code in prediction_market_agent_tooling/tools/contract.py 
263
264
265
266
def balance_of_in_tokens(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken:
    return self.balanceOf(for_address, web3=web3).as_token
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

ContractDepositableWrapperERC20BaseClass

Bases: ContractERC20BaseClass

ERC-20 standard base class extended for wrapper tokens. Although this is not a standard, it's seems to be a common pattern for wrapped tokens (at least it checks out for wxDai and wETH).

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/depositablewrapper_erc20.abi.json",
    )
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
deposit 
deposit(
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def deposit(
    self,
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send_with_value(
        api_keys=api_keys,
        function_name="deposit",
        amount_wei=amount_wei,
        tx_params=tx_params,
        web3=web3,
    )
withdraw 
withdraw(
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def withdraw(
    self,
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="withdraw",
        function_params=[amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
200
201
202
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
symbol_cached 
symbol_cached(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
204
205
206
207
208
209
210
def symbol_cached(self, web3: Web3 | None = None) -> str:
    web3 = web3 or self.get_web3()
    cache_key = create_contract_method_cache_key(self.symbol, web3)
    if cache_key not in self._cache:
        self._cache[cache_key] = self.symbol(web3=web3)
    value: str = self._cache[cache_key]
    return value
allowance 
allowance(
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
212
213
214
215
216
217
218
219
220
221
def allowance(
    self,
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei:
    allowance_for_user: int = self.call(
        "allowance", function_params=[owner, for_address], web3=web3
    )
    return Wei(allowance_for_user)
approve 
approve(
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def approve(
    self,
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="approve",
        function_params=[
            for_address,
            amount_wei,
        ],
        tx_params=tx_params,
        web3=web3,
    )
transferFrom 
transferFrom(
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def transferFrom(
    self,
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="transferFrom",
        function_params=[sender, recipient, amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
259
260
261
def balanceOf(self, for_address: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [for_address], web3=web3))
    return balance
balance_of_in_tokens 
balance_of_in_tokens(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken
Source code in prediction_market_agent_tooling/tools/contract.py 
263
264
265
266
def balance_of_in_tokens(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken:
    return self.balanceOf(for_address, web3=web3).as_token

ContractERC4626BaseClass

Bases: ContractERC20BaseClass

Class for ERC-4626, which is a superset for ERC-20.

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/erc4626.abi.json",
    )
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
asset 
asset(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
324
325
326
def asset(self, web3: Web3 | None = None) -> ChecksumAddress:
    address = self.call("asset", web3=web3)
    return Web3.to_checksum_address(address)
deposit 
deposit(
    api_keys: APIKeys,
    amount_wei: Wei,
    receiver: ChecksumAddress | None = None,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def deposit(
    self,
    api_keys: APIKeys,
    amount_wei: Wei,
    receiver: ChecksumAddress | None = None,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    receiver = receiver or api_keys.bet_from_address
    return self.send(
        api_keys=api_keys,
        function_name="deposit",
        function_params=[amount_wei, receiver],
        tx_params=tx_params,
        web3=web3,
    )
withdraw 
withdraw(
    api_keys: APIKeys,
    assets_wei: Wei,
    receiver: ChecksumAddress | None = None,
    owner: ChecksumAddress | None = None,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def withdraw(
    self,
    api_keys: APIKeys,
    assets_wei: Wei,
    receiver: ChecksumAddress | None = None,
    owner: ChecksumAddress | None = None,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    # assets_wei is the amount of assets (underlying erc20 token amount) we want to withdraw.
    receiver = receiver or api_keys.bet_from_address
    owner = owner or api_keys.bet_from_address
    return self.send(
        api_keys=api_keys,
        function_name="withdraw",
        function_params=[assets_wei, receiver, owner],
        tx_params=tx_params,
        web3=web3,
    )
withdraw_in_shares 
withdraw_in_shares(
    api_keys: APIKeys,
    shares_wei: Wei,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
365
366
367
368
369
370
def withdraw_in_shares(
    self, api_keys: APIKeys, shares_wei: Wei, web3: Web3 | None = None
) -> TxReceipt:
    # shares_wei is the amount of shares we want to withdraw.
    assets = self.convertToAssets(shares_wei, web3=web3)
    return self.withdraw(api_keys=api_keys, assets_wei=assets, web3=web3)
convertToShares 
convertToShares(
    assets: Wei, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
372
373
374
def convertToShares(self, assets: Wei, web3: Web3 | None = None) -> Wei:
    shares = Wei(self.call("convertToShares", [assets], web3=web3))
    return shares
convertToAssets 
convertToAssets(
    shares: Wei, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
376
377
378
def convertToAssets(self, shares: Wei, web3: Web3 | None = None) -> Wei:
    assets = Wei(self.call("convertToAssets", [shares], web3=web3))
    return assets
get_asset_token_contract 
get_asset_token_contract(
    web3: Web3 | None = None,
) -> (
    ContractERC20BaseClass
    | ContractDepositableWrapperERC20BaseClass
)
Source code in prediction_market_agent_tooling/tools/contract.py 
380
381
382
383
384
385
386
387
388
def get_asset_token_contract(
    self, web3: Web3 | None = None
) -> ContractERC20BaseClass | ContractDepositableWrapperERC20BaseClass:
    web3 = web3 or self.get_web3()
    contract = init_collateral_token_contract(self.asset(), web3=web3)
    assert not isinstance(
        contract, ContractERC4626BaseClass
    ), "Asset token should be either Depositable Wrapper ERC-20 or ERC-20."  # Shrinking down possible types.
    return contract
get_asset_token_balance 
get_asset_token_balance(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
390
391
392
393
394
def get_asset_token_balance(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei:
    asset_token_contract = self.get_asset_token_contract(web3=web3)
    return asset_token_contract.balanceOf(for_address, web3=web3)
deposit_asset_token 
deposit_asset_token(
    asset_value: Wei,
    api_keys: APIKeys,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def deposit_asset_token(
    self, asset_value: Wei, api_keys: APIKeys, web3: Web3 | None = None
) -> TxReceipt:
    for_address = api_keys.bet_from_address
    web3 = web3 or self.get_web3()

    asset_token_contract = self.get_asset_token_contract(web3=web3)
    # Approve vault to withdraw the erc-20 token from the user.
    asset_token_contract.approve(api_keys, self.address, asset_value, web3=web3)

    # Deposit asset token (erc20) and we will receive shares in this vault.
    receipt = self.deposit(api_keys, asset_value, for_address, web3=web3)

    return receipt
get_in_shares 
get_in_shares(amount: Wei, web3: Web3 | None = None) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
411
412
413
def get_in_shares(self, amount: Wei, web3: Web3 | None = None) -> Wei:
    # We send erc20 to the vault and receive shares in return, which can have a different value.
    return self.convertToShares(amount, web3=web3)
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
200
201
202
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
symbol_cached 
symbol_cached(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
204
205
206
207
208
209
210
def symbol_cached(self, web3: Web3 | None = None) -> str:
    web3 = web3 or self.get_web3()
    cache_key = create_contract_method_cache_key(self.symbol, web3)
    if cache_key not in self._cache:
        self._cache[cache_key] = self.symbol(web3=web3)
    value: str = self._cache[cache_key]
    return value
allowance 
allowance(
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
212
213
214
215
216
217
218
219
220
221
def allowance(
    self,
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei:
    allowance_for_user: int = self.call(
        "allowance", function_params=[owner, for_address], web3=web3
    )
    return Wei(allowance_for_user)
approve 
approve(
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def approve(
    self,
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="approve",
        function_params=[
            for_address,
            amount_wei,
        ],
        tx_params=tx_params,
        web3=web3,
    )
transferFrom 
transferFrom(
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def transferFrom(
    self,
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="transferFrom",
        function_params=[sender, recipient, amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
259
260
261
def balanceOf(self, for_address: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [for_address], web3=web3))
    return balance
balance_of_in_tokens 
balance_of_in_tokens(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken
Source code in prediction_market_agent_tooling/tools/contract.py 
263
264
265
266
def balance_of_in_tokens(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken:
    return self.balanceOf(for_address, web3=web3).as_token

OwnableContract

Bases: ContractBaseClass

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/ownable.abi.json",
    )
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
owner 
owner(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
424
425
426
def owner(self, web3: Web3 | None = None) -> ChecksumAddress:
    owner = Web3.to_checksum_address(self.call("owner", web3=web3))
    return owner
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

ContractERC721BaseClass

Bases: ContractBaseClass

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/erc721.abi.json",
    )
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
safeMint 
safeMint(
    api_keys: APIKeys,
    to_address: ChecksumAddress,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def safeMint(
    self,
    api_keys: APIKeys,
    to_address: ChecksumAddress,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="safeMint",
        function_params=[to_address],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    owner: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
452
453
454
def balanceOf(self, owner: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [owner], web3=web3))
    return balance
owner_of 
owner_of(
    token_id: int, web3: Web3 | None = None
) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
456
457
458
def owner_of(self, token_id: int, web3: Web3 | None = None) -> ChecksumAddress:
    owner = Web3.to_checksum_address(self.call("ownerOf", [token_id], web3=web3))
    return owner
name 
name(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
460
461
462
def name(self, web3: Web3 | None = None) -> str:
    name: str = self.call("name", web3=web3)
    return name
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
464
465
466
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
tokenURI 
tokenURI(tokenId: int, web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
468
469
470
def tokenURI(self, tokenId: int, web3: Web3 | None = None) -> str:
    uri: str = self.call("tokenURI", [tokenId], web3=web3)
    return uri
safeTransferFrom 
safeTransferFrom(
    api_keys: APIKeys,
    from_address: ChecksumAddress,
    to_address: ChecksumAddress,
    tokenId: int,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def safeTransferFrom(
    self,
    api_keys: APIKeys,
    from_address: ChecksumAddress,
    to_address: ChecksumAddress,
    tokenId: int,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="safeTransferFrom",
        function_params=[from_address, to_address, tokenId],
        tx_params=tx_params,
        web3=web3,
    )
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

ContractOwnableERC721BaseClass

Bases: ContractERC721BaseClass, OwnableContract

abi class-attribute instance-attribute 
abi: ABI = merge_contract_abis(
    [
        ContractERC721BaseClass(
            address=ChecksumAddress(CHECKSUM_ADDRESSS_ZERO)
        ),
        OwnableContract(
            address=ChecksumAddress(CHECKSUM_ADDRESSS_ZERO)
        ),
    ]
)
CHAIN_ID class-attribute 
CHAIN_ID: ChainID
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
owner 
owner(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
424
425
426
def owner(self, web3: Web3 | None = None) -> ChecksumAddress:
    owner = Web3.to_checksum_address(self.call("owner", web3=web3))
    return owner
safeMint 
safeMint(
    api_keys: APIKeys,
    to_address: ChecksumAddress,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def safeMint(
    self,
    api_keys: APIKeys,
    to_address: ChecksumAddress,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="safeMint",
        function_params=[to_address],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    owner: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
452
453
454
def balanceOf(self, owner: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [owner], web3=web3))
    return balance
owner_of 
owner_of(
    token_id: int, web3: Web3 | None = None
) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
456
457
458
def owner_of(self, token_id: int, web3: Web3 | None = None) -> ChecksumAddress:
    owner = Web3.to_checksum_address(self.call("ownerOf", [token_id], web3=web3))
    return owner
name 
name(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
460
461
462
def name(self, web3: Web3 | None = None) -> str:
    name: str = self.call("name", web3=web3)
    return name
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
464
465
466
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
tokenURI 
tokenURI(tokenId: int, web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
468
469
470
def tokenURI(self, tokenId: int, web3: Web3 | None = None) -> str:
    uri: str = self.call("tokenURI", [tokenId], web3=web3)
    return uri
safeTransferFrom 
safeTransferFrom(
    api_keys: APIKeys,
    from_address: ChecksumAddress,
    to_address: ChecksumAddress,
    tokenId: int,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def safeTransferFrom(
    self,
    api_keys: APIKeys,
    from_address: ChecksumAddress,
    to_address: ChecksumAddress,
    tokenId: int,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="safeTransferFrom",
        function_params=[from_address, to_address, tokenId],
        tx_params=tx_params,
        web3=web3,
    )

ContractOnGnosisChain

Bases: ContractBaseClass

Contract base class with Gnosis Chain configuration.

CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
abi instance-attribute 
abi: ABI
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

ContractProxyOnGnosisChain

Bases: ContractProxyBaseClass, ContractOnGnosisChain

Proxy contract base class with Gnosis Chain configuration.

CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/proxy.abi.json",
    )
)
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
implementation 
implementation(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
184
185
186
def implementation(self, web3: Web3 | None = None) -> ChecksumAddress:
    address = self.call("implementation", web3=web3)
    return Web3.to_checksum_address(address)

ContractERC20OnGnosisChain

Bases: ContractERC20BaseClass, ContractOnGnosisChain

ERC-20 standard base class with Gnosis Chain configuration.

CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/erc20.abi.json",
    )
)
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
200
201
202
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
symbol_cached 
symbol_cached(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
204
205
206
207
208
209
210
def symbol_cached(self, web3: Web3 | None = None) -> str:
    web3 = web3 or self.get_web3()
    cache_key = create_contract_method_cache_key(self.symbol, web3)
    if cache_key not in self._cache:
        self._cache[cache_key] = self.symbol(web3=web3)
    value: str = self._cache[cache_key]
    return value
allowance 
allowance(
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
212
213
214
215
216
217
218
219
220
221
def allowance(
    self,
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei:
    allowance_for_user: int = self.call(
        "allowance", function_params=[owner, for_address], web3=web3
    )
    return Wei(allowance_for_user)
approve 
approve(
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def approve(
    self,
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="approve",
        function_params=[
            for_address,
            amount_wei,
        ],
        tx_params=tx_params,
        web3=web3,
    )
transferFrom 
transferFrom(
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def transferFrom(
    self,
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="transferFrom",
        function_params=[sender, recipient, amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
259
260
261
def balanceOf(self, for_address: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [for_address], web3=web3))
    return balance
balance_of_in_tokens 
balance_of_in_tokens(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken
Source code in prediction_market_agent_tooling/tools/contract.py 
263
264
265
266
def balance_of_in_tokens(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken:
    return self.balanceOf(for_address, web3=web3).as_token

ContractOwnableERC721OnGnosisChain

Bases: ContractOwnableERC721BaseClass, ContractOnGnosisChain

Ownable ERC-721 standard base class with Gnosis Chain configuration.

CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
abi class-attribute instance-attribute 
abi: ABI = merge_contract_abis(
    [
        ContractERC721BaseClass(
            address=ChecksumAddress(CHECKSUM_ADDRESSS_ZERO)
        ),
        OwnableContract(
            address=ChecksumAddress(CHECKSUM_ADDRESSS_ZERO)
        ),
    ]
)
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
owner 
owner(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
424
425
426
def owner(self, web3: Web3 | None = None) -> ChecksumAddress:
    owner = Web3.to_checksum_address(self.call("owner", web3=web3))
    return owner
safeMint 
safeMint(
    api_keys: APIKeys,
    to_address: ChecksumAddress,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def safeMint(
    self,
    api_keys: APIKeys,
    to_address: ChecksumAddress,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="safeMint",
        function_params=[to_address],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    owner: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
452
453
454
def balanceOf(self, owner: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [owner], web3=web3))
    return balance
owner_of 
owner_of(
    token_id: int, web3: Web3 | None = None
) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
456
457
458
def owner_of(self, token_id: int, web3: Web3 | None = None) -> ChecksumAddress:
    owner = Web3.to_checksum_address(self.call("ownerOf", [token_id], web3=web3))
    return owner
name 
name(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
460
461
462
def name(self, web3: Web3 | None = None) -> str:
    name: str = self.call("name", web3=web3)
    return name
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
464
465
466
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
tokenURI 
tokenURI(tokenId: int, web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
468
469
470
def tokenURI(self, tokenId: int, web3: Web3 | None = None) -> str:
    uri: str = self.call("tokenURI", [tokenId], web3=web3)
    return uri
safeTransferFrom 
safeTransferFrom(
    api_keys: APIKeys,
    from_address: ChecksumAddress,
    to_address: ChecksumAddress,
    tokenId: int,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def safeTransferFrom(
    self,
    api_keys: APIKeys,
    from_address: ChecksumAddress,
    to_address: ChecksumAddress,
    tokenId: int,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="safeTransferFrom",
        function_params=[from_address, to_address, tokenId],
        tx_params=tx_params,
        web3=web3,
    )

ContractDepositableWrapperERC20OnGnosisChain

Bases: ContractDepositableWrapperERC20BaseClass, ContractERC20OnGnosisChain

Depositable Wrapper ERC-20 standard base class with Gnosis Chain configuration.

CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/depositablewrapper_erc20.abi.json",
    )
)
address instance-attribute 
address: ChecksumAddress
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
200
201
202
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
symbol_cached 
symbol_cached(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
204
205
206
207
208
209
210
def symbol_cached(self, web3: Web3 | None = None) -> str:
    web3 = web3 or self.get_web3()
    cache_key = create_contract_method_cache_key(self.symbol, web3)
    if cache_key not in self._cache:
        self._cache[cache_key] = self.symbol(web3=web3)
    value: str = self._cache[cache_key]
    return value
allowance 
allowance(
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
212
213
214
215
216
217
218
219
220
221
def allowance(
    self,
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei:
    allowance_for_user: int = self.call(
        "allowance", function_params=[owner, for_address], web3=web3
    )
    return Wei(allowance_for_user)
approve 
approve(
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def approve(
    self,
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="approve",
        function_params=[
            for_address,
            amount_wei,
        ],
        tx_params=tx_params,
        web3=web3,
    )
transferFrom 
transferFrom(
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def transferFrom(
    self,
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="transferFrom",
        function_params=[sender, recipient, amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
259
260
261
def balanceOf(self, for_address: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [for_address], web3=web3))
    return balance
balance_of_in_tokens 
balance_of_in_tokens(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken
Source code in prediction_market_agent_tooling/tools/contract.py 
263
264
265
266
def balance_of_in_tokens(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken:
    return self.balanceOf(for_address, web3=web3).as_token
deposit 
deposit(
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def deposit(
    self,
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send_with_value(
        api_keys=api_keys,
        function_name="deposit",
        amount_wei=amount_wei,
        tx_params=tx_params,
        web3=web3,
    )
withdraw 
withdraw(
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def withdraw(
    self,
    api_keys: APIKeys,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="withdraw",
        function_params=[amount_wei],
        tx_params=tx_params,
        web3=web3,
    )

ContractERC4626OnGnosisChain

Bases: ContractERC4626BaseClass, ContractERC20OnGnosisChain

ERC-4626 standard base class with Gnosis Chain configuration.

CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/erc4626.abi.json",
    )
)
address instance-attribute 
address: ChecksumAddress
get_asset_token_contract 
get_asset_token_contract(
    web3: Web3 | None = None,
) -> (
    ContractERC20OnGnosisChain
    | ContractDepositableWrapperERC20OnGnosisChain
)
Source code in prediction_market_agent_tooling/tools/contract.py 
543
544
545
546
def get_asset_token_contract(
    self, web3: Web3 | None = None
) -> ContractERC20OnGnosisChain | ContractDepositableWrapperERC20OnGnosisChain:
    return to_gnosis_chain_contract(super().get_asset_token_contract(web3=web3))
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()
symbol 
symbol(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
200
201
202
def symbol(self, web3: Web3 | None = None) -> str:
    symbol: str = self.call("symbol", web3=web3)
    return symbol
symbol_cached 
symbol_cached(web3: Web3 | None = None) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
204
205
206
207
208
209
210
def symbol_cached(self, web3: Web3 | None = None) -> str:
    web3 = web3 or self.get_web3()
    cache_key = create_contract_method_cache_key(self.symbol, web3)
    if cache_key not in self._cache:
        self._cache[cache_key] = self.symbol(web3=web3)
    value: str = self._cache[cache_key]
    return value
allowance 
allowance(
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
212
213
214
215
216
217
218
219
220
221
def allowance(
    self,
    owner: ChecksumAddress,
    for_address: ChecksumAddress,
    web3: Web3 | None = None,
) -> Wei:
    allowance_for_user: int = self.call(
        "allowance", function_params=[owner, for_address], web3=web3
    )
    return Wei(allowance_for_user)
approve 
approve(
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def approve(
    self,
    api_keys: APIKeys,
    for_address: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="approve",
        function_params=[
            for_address,
            amount_wei,
        ],
        tx_params=tx_params,
        web3=web3,
    )
transferFrom 
transferFrom(
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def transferFrom(
    self,
    api_keys: APIKeys,
    sender: ChecksumAddress,
    recipient: ChecksumAddress,
    amount_wei: Wei,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="transferFrom",
        function_params=[sender, recipient, amount_wei],
        tx_params=tx_params,
        web3=web3,
    )
balanceOf 
balanceOf(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
259
260
261
def balanceOf(self, for_address: ChecksumAddress, web3: Web3 | None = None) -> Wei:
    balance = Wei(self.call("balanceOf", [for_address], web3=web3))
    return balance
balance_of_in_tokens 
balance_of_in_tokens(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken
Source code in prediction_market_agent_tooling/tools/contract.py 
263
264
265
266
def balance_of_in_tokens(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> CollateralToken:
    return self.balanceOf(for_address, web3=web3).as_token
asset 
asset(web3: Web3 | None = None) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/contract.py 
324
325
326
def asset(self, web3: Web3 | None = None) -> ChecksumAddress:
    address = self.call("asset", web3=web3)
    return Web3.to_checksum_address(address)
deposit 
deposit(
    api_keys: APIKeys,
    amount_wei: Wei,
    receiver: ChecksumAddress | None = None,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def deposit(
    self,
    api_keys: APIKeys,
    amount_wei: Wei,
    receiver: ChecksumAddress | None = None,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    receiver = receiver or api_keys.bet_from_address
    return self.send(
        api_keys=api_keys,
        function_name="deposit",
        function_params=[amount_wei, receiver],
        tx_params=tx_params,
        web3=web3,
    )
withdraw 
withdraw(
    api_keys: APIKeys,
    assets_wei: Wei,
    receiver: ChecksumAddress | None = None,
    owner: ChecksumAddress | None = None,
    tx_params: Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def withdraw(
    self,
    api_keys: APIKeys,
    assets_wei: Wei,
    receiver: ChecksumAddress | None = None,
    owner: ChecksumAddress | None = None,
    tx_params: t.Optional[TxParams] = None,
    web3: Web3 | None = None,
) -> TxReceipt:
    # assets_wei is the amount of assets (underlying erc20 token amount) we want to withdraw.
    receiver = receiver or api_keys.bet_from_address
    owner = owner or api_keys.bet_from_address
    return self.send(
        api_keys=api_keys,
        function_name="withdraw",
        function_params=[assets_wei, receiver, owner],
        tx_params=tx_params,
        web3=web3,
    )
withdraw_in_shares 
withdraw_in_shares(
    api_keys: APIKeys,
    shares_wei: Wei,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
365
366
367
368
369
370
def withdraw_in_shares(
    self, api_keys: APIKeys, shares_wei: Wei, web3: Web3 | None = None
) -> TxReceipt:
    # shares_wei is the amount of shares we want to withdraw.
    assets = self.convertToAssets(shares_wei, web3=web3)
    return self.withdraw(api_keys=api_keys, assets_wei=assets, web3=web3)
convertToShares 
convertToShares(
    assets: Wei, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
372
373
374
def convertToShares(self, assets: Wei, web3: Web3 | None = None) -> Wei:
    shares = Wei(self.call("convertToShares", [assets], web3=web3))
    return shares
convertToAssets 
convertToAssets(
    shares: Wei, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
376
377
378
def convertToAssets(self, shares: Wei, web3: Web3 | None = None) -> Wei:
    assets = Wei(self.call("convertToAssets", [shares], web3=web3))
    return assets
get_asset_token_balance 
get_asset_token_balance(
    for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
390
391
392
393
394
def get_asset_token_balance(
    self, for_address: ChecksumAddress, web3: Web3 | None = None
) -> Wei:
    asset_token_contract = self.get_asset_token_contract(web3=web3)
    return asset_token_contract.balanceOf(for_address, web3=web3)
deposit_asset_token 
deposit_asset_token(
    asset_value: Wei,
    api_keys: APIKeys,
    web3: Web3 | None = None,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def deposit_asset_token(
    self, asset_value: Wei, api_keys: APIKeys, web3: Web3 | None = None
) -> TxReceipt:
    for_address = api_keys.bet_from_address
    web3 = web3 or self.get_web3()

    asset_token_contract = self.get_asset_token_contract(web3=web3)
    # Approve vault to withdraw the erc-20 token from the user.
    asset_token_contract.approve(api_keys, self.address, asset_value, web3=web3)

    # Deposit asset token (erc20) and we will receive shares in this vault.
    receipt = self.deposit(api_keys, asset_value, for_address, web3=web3)

    return receipt
get_in_shares 
get_in_shares(amount: Wei, web3: Web3 | None = None) -> Wei
Source code in prediction_market_agent_tooling/tools/contract.py 
411
412
413
def get_in_shares(self, amount: Wei, web3: Web3 | None = None) -> Wei:
    # We send erc20 to the vault and receive shares in return, which can have a different value.
    return self.convertToShares(amount, web3=web3)

DebuggingContract

Bases: ContractOnGnosisChain

abi class-attribute instance-attribute 
abi: ABI = abi_field_validator(
    join(
        dirname(realpath(__file__)),
        "../abis/debuggingcontract.abi.json",
    )
)
address class-attribute instance-attribute 
address: ChecksumAddress = to_checksum_address(
    "0x5Aa82E068aE6a6a1C26c42E5a59520a74Cdb8998"
)
CHAIN_ID class-attribute instance-attribute 
CHAIN_ID = chain_id
getNow 
getNow(web3: Web3 | None = None) -> int
Source code in prediction_market_agent_tooling/tools/contract.py 
561
562
563
564
565
566
567
568
569
def getNow(
    self,
    web3: Web3 | None = None,
) -> int:
    now: int = self.call(
        function_name="getNow",
        web3=web3,
    )
    return now
get_now 
get_now(web3: Web3 | None = None) -> DatetimeUTC
Source code in prediction_market_agent_tooling/tools/contract.py 
571
572
573
574
575
def get_now(
    self,
    web3: Web3 | None = None,
) -> DatetimeUTC:
    return DatetimeUTC.to_datetime_utc(self.getNow(web3))
inc 
inc(
    api_keys: APIKeys, web3: Web3 | None = None
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/contract.py 
577
578
579
580
581
582
583
584
585
586
def inc(
    self,
    api_keys: APIKeys,
    web3: Web3 | None = None,
) -> TxReceipt:
    return self.send(
        api_keys=api_keys,
        function_name="inc",
        web3=web3,
    )
merge_contract_abis classmethod 
merge_contract_abis(
    contracts: list[ContractBaseClass],
) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
77
78
79
80
81
82
@classmethod
def merge_contract_abis(cls, contracts: list["ContractBaseClass"]) -> ABI:
    merged_abi: list[dict[t.Any, t.Any]] = sum(
        (json.loads(contract.abi) for contract in contracts), []
    )
    return abi_field_validator(json.dumps(merged_abi))
get_web3_contract 
get_web3_contract(web3: Web3 | None = None) -> Web3Contract
Source code in prediction_market_agent_tooling/tools/contract.py 
84
85
86
def get_web3_contract(self, web3: Web3 | None = None) -> Web3Contract:
    web3 = web3 or self.get_web3()
    return web3.eth.contract(address=self.address, abi=self.abi)
call 
call(
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    web3: Web3 | None = None,
) -> t.Any

Used for reading from the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def call(
    self,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    web3: Web3 | None = None,
) -> t.Any:
    """
    Used for reading from the contract.
    """
    web3 = web3 or self.get_web3()
    return call_function_on_contract(
        web3=web3,
        contract_address=self.address,
        contract_abi=self.abi,
        function_name=function_name,
        function_params=function_params,
    )
send 
send(
    api_keys: APIKeys,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract.

Source code in prediction_market_agent_tooling/tools/contract.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send(
    self,
    api_keys: APIKeys,
    function_name: str,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract.
    """

    if api_keys.safe_address_checksum:
        return send_function_on_contract_tx_using_safe(
            web3=web3 or self.get_web3(),
            contract_address=self.address,
            contract_abi=self.abi,
            from_private_key=api_keys.bet_from_private_key,
            safe_address=api_keys.safe_address_checksum,
            function_name=function_name,
            function_params=function_params,
            tx_params=tx_params,
            timeout=timeout,
        )
    return send_function_on_contract_tx(
        web3=web3 or self.get_web3(),
        contract_address=self.address,
        contract_abi=self.abi,
        from_private_key=api_keys.bet_from_private_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
        timeout=timeout,
    )
send_with_value 
send_with_value(
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt

Used for changing a state (writing) to the contract, including sending chain's native currency.

Source code in prediction_market_agent_tooling/tools/contract.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def send_with_value(
    self,
    api_keys: APIKeys,
    function_name: str,
    amount_wei: Wei,
    function_params: t.Optional[list[t.Any] | dict[str, t.Any]] = None,
    tx_params: t.Optional[TxParams] = None,
    timeout: int = 180,
    web3: Web3 | None = None,
) -> TxReceipt:
    """
    Used for changing a state (writing) to the contract, including sending chain's native currency.
    """
    return self.send(
        api_keys=api_keys,
        function_name=function_name,
        function_params=function_params,
        tx_params={"value": amount_wei.value, **(tx_params or {})},
        timeout=timeout,
        web3=web3,
    )
get_transaction_count classmethod 
get_transaction_count(
    for_address: ChecksumAddress,
) -> Nonce
Source code in prediction_market_agent_tooling/tools/contract.py 
164
165
166
@classmethod
def get_transaction_count(cls, for_address: ChecksumAddress) -> Nonce:
    return cls.get_web3().eth.get_transaction_count(for_address)
get_web3 classmethod 
get_web3() -> Web3
Source code in prediction_market_agent_tooling/tools/contract.py 
168
169
170
@classmethod
def get_web3(cls) -> Web3:
    return RPCConfig().get_web3()

abi_field_validator 

abi_field_validator(value: str) -> ABI
Source code in prediction_market_agent_tooling/tools/contract.py 
34
35
36
37
38
39
40
41
42
43
def abi_field_validator(value: str) -> ABI:
    if value.endswith(".json"):
        with open(value) as f:
            value = f.read()

    try:
        json.loads(value)  # Test if it's valid JSON content.
        return ABI(value)
    except json.decoder.JSONDecodeError:
        raise ValueError(f"Invalid ABI: {value}")

wait_until_nonce_changed 

wait_until_nonce_changed(
    for_address: ChecksumAddress,
    timeout: int = 10,
    sleep_time: int = 1,
) -> t.Generator[None, None, None]
Source code in prediction_market_agent_tooling/tools/contract.py 
46
47
48
49
50
51
52
53
54
55
56
57
@contextmanager
def wait_until_nonce_changed(
    for_address: ChecksumAddress, timeout: int = 10, sleep_time: int = 1
) -> t.Generator[None, None, None]:
    current_nonce = ContractOnGnosisChain.get_transaction_count(for_address)
    yield
    start_monotonic = time.monotonic()
    while (
        time.monotonic() - start_monotonic < timeout
        and current_nonce == ContractOnGnosisChain.get_transaction_count(for_address)
    ):
        time.sleep(sleep_time)

contract_implements_function 

contract_implements_function(
    contract_address: ChecksumAddress,
    function_name: str,
    web3: Web3,
    function_arg_types: list[str] | None = None,
    look_for_proxy_contract: bool = True,
) -> bool
Source code in prediction_market_agent_tooling/tools/contract.py 
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
def contract_implements_function(
    contract_address: ChecksumAddress,
    function_name: str,
    web3: Web3,
    function_arg_types: list[str] | None = None,
    look_for_proxy_contract: bool = True,
) -> bool:
    function_signature = f"{function_name}({','.join(function_arg_types or [])})"
    function_selector = web3.keccak(text=function_signature)[0:4].hex()[2:]
    # 1. Check directly in bytecode
    bytecode = web3.eth.get_code(contract_address).hex()
    if function_selector in bytecode:
        return True
    contract_code = web3.eth.get_code(contract_address).hex()
    implements = function_selector in contract_code

    # If not found directly and we should check proxies
    if not implements and look_for_proxy_contract:
        # Case 1: Check if it's a standard proxy (has implementation() function)
        if contract_implements_function(
            contract_address, "implementation", web3, look_for_proxy_contract=False
        ):
            # Get the implementation address and check the function there
            implementation_address = ContractProxyOnGnosisChain(
                address=contract_address
            ).implementation()
            implements = contract_implements_function(
                implementation_address,
                function_name=function_name,
                web3=web3,
                function_arg_types=function_arg_types,
                look_for_proxy_contract=False,
            )
        else:
            # Case 2: Check if it's a minimal proxy contract
            implements = minimal_proxy_implements_function(
                contract_address=contract_address,
                function_name=function_name,
                web3=web3,
                function_arg_types=function_arg_types,
            )

    return implements

minimal_proxy_implements_function 

minimal_proxy_implements_function(
    contract_address: ChecksumAddress,
    function_name: str,
    web3: Web3,
    function_arg_types: list[str] | None = None,
) -> bool
Source code in prediction_market_agent_tooling/tools/contract.py 
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
def minimal_proxy_implements_function(
    contract_address: ChecksumAddress,
    function_name: str,
    web3: Web3,
    function_arg_types: list[str] | None = None,
) -> bool:
    try:
        # Read storage slot 0 which should contain the implementation address in minimal proxies
        raw_slot_0 = web3.eth.get_storage_at(contract_address, 0)
        singleton_address = eth_abi.decode(["address"], raw_slot_0)[0]
        # Recurse into singleton
        return contract_implements_function(
            Web3.to_checksum_address(singleton_address),
            function_name=function_name,
            web3=web3,
            function_arg_types=function_arg_types,
            look_for_proxy_contract=False,
        )
    except DecodingError:
        logger.info(f"Error decoding contract address for singleton")
        return False

init_collateral_token_contract 

init_collateral_token_contract(
    address: ChecksumAddress, web3: Web3 | None
) -> ContractERC20BaseClass

Checks if the given contract is Depositable ERC-20, ERC-20 or ERC-4626 and returns the appropriate class instance. Throws an error if the contract is neither of them.

Source code in prediction_market_agent_tooling/tools/contract.py 
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
def init_collateral_token_contract(
    address: ChecksumAddress, web3: Web3 | None
) -> ContractERC20BaseClass:
    """
    Checks if the given contract is Depositable ERC-20, ERC-20 or ERC-4626 and returns the appropriate class instance.
    Throws an error if the contract is neither of them.
    """
    web3 = web3 or RPCConfig().get_web3()

    if contract_implements_function(address, "asset", web3=web3):
        return ContractERC4626BaseClass(address=address)

    elif contract_implements_function(
        address,
        "deposit",
        web3=web3,
    ):
        return ContractDepositableWrapperERC20BaseClass(address=address)

    elif contract_implements_function(
        address,
        "balanceOf",
        web3=web3,
        function_arg_types=["address"],
    ):
        return ContractERC20BaseClass(address=address)

    else:
        raise ValueError(
            f"Contract at {address} is neither Depositable ERC-20, ERC-20 nor ERC-4626."
        )

to_gnosis_chain_contract 

to_gnosis_chain_contract(
    contract: ContractERC20BaseClass,
) -> ContractERC20OnGnosisChain
Source code in prediction_market_agent_tooling/tools/contract.py 
690
691
692
693
694
695
696
697
698
699
700
def to_gnosis_chain_contract(
    contract: ContractERC20BaseClass,
) -> ContractERC20OnGnosisChain:
    if isinstance(contract, ContractERC4626BaseClass):
        return ContractERC4626OnGnosisChain(address=contract.address)
    elif isinstance(contract, ContractDepositableWrapperERC20BaseClass):
        return ContractDepositableWrapperERC20OnGnosisChain(address=contract.address)
    elif isinstance(contract, ContractERC20BaseClass):
        return ContractERC20OnGnosisChain(address=contract.address)
    else:
        raise ValueError("Unsupported contract type")

create_contract_method_cache_key 

create_contract_method_cache_key(
    method: Callable[[Any], Any], web3: Web3
) -> str
Source code in prediction_market_agent_tooling/tools/contract.py 
703
704
705
706
def create_contract_method_cache_key(
    method: t.Callable[[t.Any], t.Any], web3: Web3
) -> str:
    return f"{method.__name__}-{str(web3.provider)}"

costs

Costs

Bases: BaseModel

time instance-attribute 
time: float
cost instance-attribute 
cost: float

openai_costs 

openai_costs(
    model: str | None = None,
) -> t.Generator[Costs, None, None]
Source code in prediction_market_agent_tooling/tools/costs.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@contextmanager
def openai_costs(model: str | None = None) -> t.Generator[Costs, None, None]:
    costs = Costs(time=0, cost=0)
    start_time = time()

    with get_openai_callback() as cb:
        yield costs
        if cb.total_tokens > 0 and cb.total_cost == 0 and model is not None:
            # TODO: this is a hack to get the cost for an unsupported model
            cb.total_cost = get_llm_api_call_cost(
                model=model,
                prompt_tokens=cb.prompt_tokens,
                completion_tokens=cb.completion_tokens,
            )
        costs.time = time() - start_time
        costs.cost = cb.total_cost

custom_exceptions

CantPayForGasError

Bases: ValueError

OutOfFundsError

Bases: ValueError

datetime_utc

DatetimeUTC

Bases: datetime

As a subclass of datetime instead of NewType because otherwise it doesn't work with issubclass command which is required for SQLModel/Pydantic.

from_datetime staticmethod 
from_datetime(dt: datetime) -> DatetimeUTC

Converts a datetime object to DatetimeUTC, ensuring it is timezone-aware in UTC.

Source code in prediction_market_agent_tooling/tools/datetime_utc.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@staticmethod
def from_datetime(dt: datetime) -> "DatetimeUTC":
    """
    Converts a datetime object to DatetimeUTC, ensuring it is timezone-aware in UTC.
    """
    if dt.tzinfo is None:
        logger.warning(
            f"tzinfo not provided, assuming the timezone of {dt=} is UTC."
        )
        dt = dt.replace(tzinfo=pytz.UTC)
    else:
        dt = dt.astimezone(pytz.UTC)
    return DatetimeUTC(
        dt.year,
        dt.month,
        dt.day,
        dt.hour,
        dt.minute,
        dt.second,
        dt.microsecond,
        tzinfo=pytz.UTC,
    )
to_datetime_utc staticmethod 
to_datetime_utc(value: datetime | int | str) -> DatetimeUTC
Source code in prediction_market_agent_tooling/tools/datetime_utc.py 
63
64
65
66
67
68
69
70
71
72
73
74
@staticmethod
def to_datetime_utc(value: datetime | int | str) -> "DatetimeUTC":
    if isinstance(value, int):
        # Divide by 1000 if the timestamp is assumed to be in milliseconds (if not, 1e11 would be year 5138).
        value = int(value / 1000) if value > 1e11 else value
        # In the past, we had bugged data where timestamp was huge and Python errored out.
        max_timestamp = int((datetime.max - timedelta(days=1)).timestamp())
        value = min(value, max_timestamp)
        value = datetime.fromtimestamp(value, tz=pytz.UTC)
    elif isinstance(value, str):
        value = parser.parse(value)
    return DatetimeUTC.from_datetime(value)

db

db_manager

DBManager 
DBManager(sqlalchemy_db_url: str | None = None)
Source code in prediction_market_agent_tooling/tools/db/db_manager.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, sqlalchemy_db_url: str | None = None) -> None:
    if hasattr(self, "_initialized"):
        return
    sqlalchemy_db_url = (
        sqlalchemy_db_url or APIKeys().sqlalchemy_db_url.get_secret_value()
    )
    self._engine = create_engine(
        sqlalchemy_db_url,
        json_serializer=json_serializer,
        json_deserializer=json_deserializer,
        pool_size=2,
    )
    self.cache_table_initialized: dict[str, bool] = {}
    self._initialized = True
cache_table_initialized instance-attribute 
cache_table_initialized: dict[str, bool] = {}
get_session 
get_session() -> Generator[Session, None, None]
Source code in prediction_market_agent_tooling/tools/db/db_manager.py 
45
46
47
48
@contextmanager
def get_session(self) -> Generator[Session, None, None]:
    with Session(self._engine) as session:
        yield session
get_connection 
get_connection() -> Generator[Connection, None, None]
Source code in prediction_market_agent_tooling/tools/db/db_manager.py 
50
51
52
53
@contextmanager
def get_connection(self) -> Generator[Connection, None, None]:
    with self.get_session() as session:
        yield session.connection()
create_tables 
create_tables(
    sqlmodel_tables: Sequence[type[SQLModel]] | None = None,
) -> None
Source code in prediction_market_agent_tooling/tools/db/db_manager.py 
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
def create_tables(
    self, sqlmodel_tables: Sequence[type[SQLModel]] | None = None
) -> None:
    if sqlmodel_tables is not None:
        tables_to_create = []
        for sqlmodel_table in sqlmodel_tables:
            table_name = (
                sqlmodel_table.__tablename__()
                if callable(sqlmodel_table.__tablename__)
                else sqlmodel_table.__tablename__
            )
            table = SQLModel.metadata.tables[table_name]
            if not self.cache_table_initialized.get(table.name):
                tables_to_create.append(table)
    else:
        tables_to_create = None

    # Create tables in the database
    if tables_to_create is None or len(tables_to_create) > 0:
        with self.get_connection() as connection:
            SQLModel.metadata.create_all(connection, tables=tables_to_create)
            connection.commit()

    # Update cache to mark tables as initialized
    if tables_to_create:
        for table in tables_to_create:
            self.cache_table_initialized[table.name] = True

google_utils

search_google_gcp 

search_google_gcp(
    query: str | None = None,
    num: int = 3,
    exact_terms: str | None = None,
    exclude_terms: str | None = None,
    link_site: str | None = None,
    site_search: str | None = None,
    site_search_filter: Literal["e", "i"] | None = None,
) -> list[str]

Search Google using a custom search engine.

Source code in prediction_market_agent_tooling/tools/google_utils.py 
13
14
15
16
17
18
19
20
21
22
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
@tenacity.retry(
    wait=tenacity.wait_fixed(1),
    stop=tenacity.stop_after_attempt(3),
    after=lambda x: logger.debug(f"search_google_gcp failed, {x.attempt_number=}."),
)
@db_cache(max_age=timedelta(days=1))
def search_google_gcp(
    query: str | None = None,
    num: int = 3,
    exact_terms: str | None = None,
    exclude_terms: str | None = None,
    link_site: str | None = None,
    site_search: str | None = None,
    site_search_filter: t.Literal["e", "i"] | None = None,
) -> list[str]:
    """Search Google using a custom search engine."""
    keys = APIKeys()
    service = build(
        "customsearch", "v1", developerKey=keys.google_search_api_key.get_secret_value()
    )
    # See https://developers.google.com/custom-search/v1/reference/rest/v1/cse/list
    params: dict[str, str | int | None] = dict(
        q=query,
        cx=keys.google_search_engine_id.get_secret_value(),
        num=num,
        exactTerms=exact_terms,
        excludeTerms=exclude_terms,
        linkSite=link_site,
        siteSearch=site_search,
        siteSearchFilter=site_search_filter,
    )
    params_without_optional = {k: v for k, v in params.items() if v is not None}
    search = service.cse().list(**params_without_optional).execute()

    try:
        return (
            [result["link"] for result in search["items"]]
            if int(search["searchInformation"]["totalResults"]) > 0
            else []
        )
    except KeyError as e:
        raise ValueError(f"Can not parse results: {search}") from e

search_google_serper 

search_google_serper(q: str) -> list[str]

Search Google using the Serper API.

Source code in prediction_market_agent_tooling/tools/google_utils.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@tenacity.retry(
    wait=tenacity.wait_fixed(1),
    stop=tenacity.stop_after_attempt(3),
    after=lambda x: logger.debug(f"search_google_serper failed, {x.attempt_number=}."),
)
@db_cache(max_age=timedelta(days=1))
def search_google_serper(q: str) -> list[str]:
    """Search Google using the Serper API."""
    url = "https://google.serper.dev/search"
    response = requests.post(
        url,
        headers={
            "X-API-KEY": APIKeys().serper_api_key.get_secret_value(),
            "Content-Type": "application/json",
        },
        json={"q": q},
    )
    response.raise_for_status()
    parsed = response.json()
    return [x["link"] for x in parsed["organic"] if x.get("link")]

hexbytes_custom

hex_serializer module-attribute 

hex_serializer = plain_serializer_function_ser_schema(
    function=lambda x: hex()
)

BaseHex

schema_pattern class-attribute 
schema_pattern: str = '^0x([0-9a-f][0-9a-f])*$'
schema_examples class-attribute 
schema_examples: tuple[str, ...] = (
    "0x",
    "0xd4",
    "0xd4e5",
    "0xd4e56740",
    "0xd4e56740f876aef8",
    "0xd4e56740f876aef8c010b86a40d5f567",
    "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3",
)

HexBytes

Bases: HexBytes, BaseHex

Use when receiving hexbytes.HexBytes values. Includes a pydantic validator and serializer.

schema_pattern class-attribute 
schema_pattern: str = '^0x([0-9a-f][0-9a-f])*$'
schema_examples class-attribute 
schema_examples: tuple[str, ...] = (
    "0x",
    "0xd4",
    "0xd4e5",
    "0xd4e56740",
    "0xd4e56740f876aef8",
    "0xd4e56740f876aef8c010b86a40d5f567",
    "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3",
)
fromhex classmethod 
fromhex(hex_str: str) -> HexBytes
Source code in prediction_market_agent_tooling/tools/hexbytes_custom.py 
58
59
60
61
@classmethod
def fromhex(cls, hex_str: str) -> "HexBytes":
    value = hex_str[2:] if hex_str.startswith("0x") else hex_str
    return super().fromhex(value)
as_int 
as_int() -> int
Source code in prediction_market_agent_tooling/tools/hexbytes_custom.py 
69
70
def as_int(self) -> int:
    return int(self.hex(), 16)

httpx_cached_client

HttpxCachedClient 

HttpxCachedClient()
Source code in prediction_market_agent_tooling/tools/httpx_cached_client.py 
 5
 6
 7
 8
 9
10
11
def __init__(self) -> None:
    storage = hishel.FileStorage(
        ttl=24 * 60 * 60,
        check_ttl_every=1 * 60 * 60,
    )
    controller = hishel.Controller(force_cache=True)
    self.client = hishel.CacheClient(storage=storage, controller=controller)
client instance-attribute 
client = CacheClient(storage=storage, controller=controller)
get_client 
get_client() -> hishel.CacheClient
Source code in prediction_market_agent_tooling/tools/httpx_cached_client.py 
13
14
def get_client(self) -> hishel.CacheClient:
    return self.client

image_gen

image_gen

generate_image 
generate_image(
    prompt: str,
    model: str = "dall-e-3",
    size: Literal[
        "256x256",
        "512x512",
        "1024x1024",
        "1792x1024",
        "1024x1792",
    ] = "1024x1024",
    quality: Literal["standard", "hd"] = "standard",
) -> ImageType
Source code in prediction_market_agent_tooling/tools/image_gen/image_gen.py 
12
13
14
15
16
17
18
19
20
21
22
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
def generate_image(
    prompt: str,
    model: str = "dall-e-3",
    size: t.Literal[
        "256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"
    ] = "1024x1024",
    quality: t.Literal["standard", "hd"] = "standard",
) -> ImageType:
    try:
        from openai import OpenAI
    except ImportError:
        raise ImportError(
            "openai not installed, please install extras `openai` to use this function."
        )
    response = OpenAI(
        api_key=APIKeys().openai_api_key.get_secret_value(),
    ).images.generate(
        model=model,
        prompt=prompt,
        size=size,
        quality=quality,
        response_format="b64_json",
        n=1,
    )

    if response.data is None or len(response.data) == 0:
        raise ValueError("No image data returned from the API.")

    image_data = response.data[0]

    image = Image.open(
        io.BytesIO(
            base64.b64decode(
                check_not_none(
                    image_data.b64_json, "Can't be none if response_format is b64_json."
                )
            )
        )
    )
    return image

market_thumbnail_gen

rewrite_question_into_image_generation_prompt 
rewrite_question_into_image_generation_prompt(
    question: str,
) -> str
Source code in prediction_market_agent_tooling/tools/image_gen/market_thumbnail_gen.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@observe()
def rewrite_question_into_image_generation_prompt(question: str) -> str:
    try:
        from langchain_openai import ChatOpenAI
    except ImportError:
        raise ImportError(
            "openai not installed, please install extras `langchain` to use this function."
        )
    llm = ChatOpenAI(
        model_name="gpt-4o-2024-08-06",
        temperature=0.0,
        openai_api_key=APIKeys().openai_api_key,
    )
    rewritten = str(
        llm.invoke(
            f"Rewrite this prediction market question '{question}' into a form that will generate nice thumbnail with DALL-E-3."
            "The thumbnail should be catchy and visually appealing. With a large object in the center of the image.",
            config=get_langfuse_langchain_config(),
        ).content
    )
    return rewritten
generate_image_for_market 
generate_image_for_market(question: str) -> ImageType
Source code in prediction_market_agent_tooling/tools/image_gen/market_thumbnail_gen.py 
34
35
36
def generate_image_for_market(question: str) -> ImageType:
    prompt = rewrite_question_into_image_generation_prompt(question)
    return generate_image(prompt)

ipfs

ipfs_handler

IPFSHandler 
IPFSHandler(api_keys: APIKeys)
Source code in prediction_market_agent_tooling/tools/ipfs/ipfs_handler.py 
12
13
14
15
16
def __init__(self, api_keys: APIKeys):
    self.pinata = PinataPy(
        api_keys.pinata_api_key.get_secret_value(),
        api_keys.pinata_api_secret.get_secret_value(),
    )
pinata instance-attribute 
pinata = PinataPy(get_secret_value(), get_secret_value())
upload_file 
upload_file(file_path: str) -> IPFSCIDVersion0
Source code in prediction_market_agent_tooling/tools/ipfs/ipfs_handler.py 
18
19
20
21
22
23
def upload_file(self, file_path: str) -> IPFSCIDVersion0:
    return IPFSCIDVersion0(
        self.pinata.pin_file_to_ipfs(file_path, save_absolute_paths=False)[
            "IpfsHash"
        ]
    )
store_agent_result 
store_agent_result(
    agent_result: IPFSAgentResult,
) -> IPFSCIDVersion0
Source code in prediction_market_agent_tooling/tools/ipfs/ipfs_handler.py 
25
26
27
28
29
30
def store_agent_result(self, agent_result: IPFSAgentResult) -> IPFSCIDVersion0:
    with tempfile.NamedTemporaryFile(mode="r+", encoding="utf-8") as json_file:
        json.dump(agent_result.model_dump(), json_file, indent=4)
        json_file.flush()
        ipfs_hash = self.upload_file(json_file.name)
        return ipfs_hash
unpin_file 
unpin_file(hash_to_remove: str) -> None
Source code in prediction_market_agent_tooling/tools/ipfs/ipfs_handler.py 
32
33
def unpin_file(self, hash_to_remove: str) -> None:
    self.pinata.remove_pin_from_ipfs(hash_to_remove=hash_to_remove)

is_invalid

QUESTION_IS_INVALID_PROMPT module-attribute 

QUESTION_IS_INVALID_PROMPT = 'Main signs about an invalid question (sometimes referred to as a "market"):\n- The market\'s question is about immoral violence, death or assassination.\n- The violent event can be caused by a single conscious being.\n- The violent event is done illegally.\n- The market should not directly incentivize immoral violent (such as murder, rape or unjust imprisonment) actions which could likely be performed by any participant.\n- Invalid: Will Donald Trump be alive on the 01/12/2021? (Anyone could bet on "No" and kill him for a guaranteed profit. Anyone could bet on "Yes" to effectively put a bounty on his head).\n- Invalid: Will Hera be a victim of swatting in 2020? (Anyone could falsely call the emergency services on him in order to win the bet)\n- This does not prevent markets:\n  - Whose topics are violent events not caused by conscious beings.\n  - Valid: How many people will die from COVID19 in 2020? (Viruses don’t use prediction markets).\n  - Whose main source of uncertainty is not related to a potential violent action.\n  - Valid: Will Trump win the 2020 US presidential election? (The main source of uncertainty is the vote of US citizens, not a potential murder of a presidential candidate).\n  - Which could give an incentive only to specific participants to commit an immoral violent action, but are in practice unlikely.\n  - Valid: Will the US be engaged in a military conflict with a UN member state in 2021? (It’s unlikely for the US to declare war in order to win a bet on this market).\n  - Valid: Will Derek Chauvin go to jail for the murder of George Flyod? (It’s unlikely that the jurors would collude to make a wrong verdict in order to win this market).\n- Questions with relative dates will resolve as invalid. Dates must be stated in absolute terms, not relative depending on the current time. But they can be relative to the event specified in the question itself.\n- Invalid: Who will be the president of the United States in 6 months? ("in 6 months depends on the current time").\n- Invalid: In the next 14 days, will Gnosis Chain gain another 1M users? ("in the next 14 days depends on the current time").\n- Valid: Will GNO price go up 10 days after Gnosis Pay cashback program is announced? ("10 days after" is relative to the event in the question, so we can determine absolute value).\n- Questions about moral values and not facts will be resolved as invalid.\n- Invalid: "Is it ethical to eat meat?".\n\nFollow a chain of thought to evaluate if the question is invalid:\n\nFirst, write the parts of the following question:\n\n"{question}"\n\nThen, write down what is the future event of the question, what it refers to and when that event will happen if the question contains it.\n\nThen, explain why do you think it is or isn\'t invalid.\n\nFinally, write your final decision, write `decision: ` followed by either "yes it is invalid" or "no it isn\'t invalid" about the question. Don\'t write anything else after that. You must include "yes" or "no".\n'

is_invalid 

is_invalid(
    question: str,
    engine: str = "gpt-4o-2024-08-06",
    temperature: float = LLM_SUPER_LOW_TEMPERATURE,
    seed: int = LLM_SEED,
    prompt_template: str = QUESTION_IS_INVALID_PROMPT,
    max_tokens: int = 1024,
) -> bool

Evaluate if the question is actually answerable.

Source code in prediction_market_agent_tooling/tools/is_invalid.py 
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
@tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_fixed(1))
@observe()
@db_cache
def is_invalid(
    question: str,
    engine: str = "gpt-4o-2024-08-06",
    temperature: float = LLM_SUPER_LOW_TEMPERATURE,
    seed: int = LLM_SEED,
    prompt_template: str = QUESTION_IS_INVALID_PROMPT,
    max_tokens: int = 1024,
) -> bool:
    """
    Evaluate if the question is actually answerable.
    """
    try:
        from langchain.prompts import ChatPromptTemplate
        from langchain_openai import ChatOpenAI
    except ImportError:
        logger.error("langchain not installed, skipping is_invalid")
        return True

    llm = ChatOpenAI(
        model_name=engine,
        temperature=temperature,
        seed=seed,
        openai_api_key=APIKeys().openai_api_key,
    )

    prompt = ChatPromptTemplate.from_template(template=prompt_template)
    messages = prompt.format_messages(question=question)
    completion = str(
        llm.invoke(
            messages, max_tokens=max_tokens, config=get_langfuse_langchain_config()
        ).content
    )

    return parse_decision_yes_no_completion(question, completion)

is_predictable

QUESTION_IS_PREDICTABLE_BINARY_PROMPT module-attribute 

QUESTION_IS_PREDICTABLE_BINARY_PROMPT = "Main signs about a fully qualified question (sometimes referred to as a \"market\"):\n- The market's question needs to be specific, without use of pronouns.\n- The market must refer to a future event — not something that already happened. This is a hard requirement. If the event has already occurred, the question is not fully qualified, even if it is specific and answerable.\n- The market's question needs to have a clear time frame.\n- The event in the market's question doesn't have to be ultra-specific, it will be decided by a crowd later on.\n- If the market's question contains date, but without an year, it's okay.\n- If the market's question contains year, but without an exact date, it's okay.\n- The market's question can not be about itself or refer to itself.\n- The answer is probably Google-able, after the event happened.\n\nFollow a chain of thought to evaluate if the question is fully qualified:\n\nFirst, write the parts of the following question:\n\n\"{question}\"\n\nThen, write down what is the future event of the question, what it refers to and when that event will happen if the question contains it.\n\nThen, explain why do you think it is or isn't fully qualified.\n\nFinally, write your final decision, write `decision: ` followed by either \"yes it is fully qualified\" or \"no it isn't fully qualified\" about the question. Don't write anything else after that. You must include \"yes\" or \"no\".\n"

QUESTION_IS_PREDICTABLE_WITHOUT_DESCRIPTION_PROMPT module-attribute 

QUESTION_IS_PREDICTABLE_WITHOUT_DESCRIPTION_PROMPT = 'Main signs about a fully self-contained question (sometimes referred to as a "market"):\n- Description of the question can not contain any additional information required to answer the question.\n\nFor the question:\n\n```\n{question}\n```\n\nAnd the description:\n\n```\n{description}\n```\n\nDescription refers only to the text above and nothing else. \n\nEven if the question is somewhat vague, but even the description does not contain enough of extra information, it\'s okay and the question is fully self-contained. \nIf the question is vague and the description contains the information required to answer the question, it\'s not fully self-contained and the answer is "no".\n\nFollow a chain of thought to evaluate if the question doesn\'t need the description to be answered.\n\nStart by examining the question and the description in detail. Write down their parts, what they refer to and what they contain. \n\nContinue by writing comparison of the question and the description content. Write down what the question contains and what the description contains.\n\nExplain, why do you think it does or doesn\'t need the description.\n\nDescription can contain additional information, but it can not contain any information required to answer the question.\n\nDescription can contain additional information about the exact resolution criteria, but the question should be answerable even without it.\n\nAs long as the question contains some time frame, it\'s okay if the description only specifies it in more detail.\n\nDescription usually contains the question in more detailed form, but the question on its own should be answerable.\n\nFor example, that means, description can not contain date if question doesn\'t contain it. Description can not contain target if the question doesn\'t contain it, etc.\n\nFinally, write your final decision, write `decision: ` followed by either "yes it is fully self-contained" or "no it isn\'t fully self-contained" about the question. Don\'t write anything else after that. You must include "yes" or "no".\n'

is_predictable_binary 

is_predictable_binary(
    question: str,
    engine: str = "gpt-4o-2024-08-06",
    prompt_template: str = QUESTION_IS_PREDICTABLE_BINARY_PROMPT,
    max_tokens: int = 1024,
) -> bool

Evaluate if the question is actually answerable.

Source code in prediction_market_agent_tooling/tools/is_predictable.py 
 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
109
110
111
112
113
114
115
@tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_fixed(1))
@observe()
@db_cache
def is_predictable_binary(
    question: str,
    engine: str = "gpt-4o-2024-08-06",
    prompt_template: str = QUESTION_IS_PREDICTABLE_BINARY_PROMPT,
    max_tokens: int = 1024,
) -> bool:
    """
    Evaluate if the question is actually answerable.
    """
    try:
        from langchain.prompts import ChatPromptTemplate
        from langchain_openai import ChatOpenAI
    except ImportError:
        logger.error("langchain not installed, skipping is_predictable_binary")
        return True

    llm = ChatOpenAI(
        model_name=engine,
        temperature=LLM_SUPER_LOW_TEMPERATURE,
        seed=LLM_SEED,
        openai_api_key=APIKeys().openai_api_key,
    )

    prompt = ChatPromptTemplate.from_template(template=prompt_template)
    messages = prompt.format_messages(question=question)
    completion = str(
        llm.invoke(
            messages, max_tokens=max_tokens, config=get_langfuse_langchain_config()
        ).content
    )

    return parse_decision_yes_no_completion(question, completion)

is_predictable_without_description 

is_predictable_without_description(
    question: str,
    description: str,
    engine: str = "gpt-4o-2024-08-06",
    prompt_template: str = QUESTION_IS_PREDICTABLE_WITHOUT_DESCRIPTION_PROMPT,
    max_tokens: int = 1024,
) -> bool

Evaluate if the question is fully self-contained.

Source code in prediction_market_agent_tooling/tools/is_predictable.py 
118
119
120
121
122
123
124
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
@tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_fixed(1))
@observe()
@db_cache
def is_predictable_without_description(
    question: str,
    description: str,
    engine: str = "gpt-4o-2024-08-06",
    prompt_template: str = QUESTION_IS_PREDICTABLE_WITHOUT_DESCRIPTION_PROMPT,
    max_tokens: int = 1024,
) -> bool:
    """
    Evaluate if the question is fully self-contained.
    """
    try:
        from langchain.prompts import ChatPromptTemplate
        from langchain_openai import ChatOpenAI
    except ImportError:
        logger.error(
            "langchain not installed, skipping is_predictable_without_description"
        )
        return True

    llm = ChatOpenAI(
        model_name=engine,
        temperature=LLM_SUPER_LOW_TEMPERATURE,
        seed=LLM_SEED,
        openai_api_key=APIKeys().openai_api_key,
    )

    prompt = ChatPromptTemplate.from_template(template=prompt_template)
    messages = prompt.format_messages(
        question=question,
        description=description,
    )
    completion = str(
        llm.invoke(
            messages, max_tokens=max_tokens, config=get_langfuse_langchain_config()
        ).content
    )

    return parse_decision_yes_no_completion(question, completion)

parse_decision_yes_no_completion 

parse_decision_yes_no_completion(
    question: str, completion: str
) -> bool
Source code in prediction_market_agent_tooling/tools/is_predictable.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
def parse_decision_yes_no_completion(question: str, completion: str) -> bool:
    logger.debug(completion)
    try:
        decision = completion.lower().rsplit("decision", 1)[1]
    except IndexError as e:
        raise ValueError(f"Invalid completion for `{question}`: {completion}") from e

    if "yes" in decision:
        return True
    elif "no" in decision:
        return False
    else:
        raise ValueError(f"Invalid completion for `{question}`: {completion}")

langfuse_

P module-attribute 

P = ParamSpec('P')

R module-attribute 

R = TypeVar('R')

observe 

observe(
    name: Optional[str] = None,
    as_type: Optional[Literal["generation"]] = None,
    capture_input: bool = True,
    capture_output: bool = True,
    transform_to_string: Optional[
        Callable[[Iterable[Any]], str]
    ] = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]
Source code in prediction_market_agent_tooling/tools/langfuse_.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def observe(
    name: Optional[str] = None,
    as_type: Optional[Literal["generation"]] = None,
    capture_input: bool = True,
    capture_output: bool = True,
    transform_to_string: Optional[Callable[[Iterable[Any]], str]] = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    casted: Callable[[Callable[P, R]], Callable[P, R]] = original_observe(
        name=name,
        as_type=as_type,
        capture_input=capture_input,
        capture_output=capture_output,
        transform_to_string=transform_to_string,
    )
    return casted

get_langfuse_langchain_config 

get_langfuse_langchain_config() -> RunnableConfig
Source code in prediction_market_agent_tooling/tools/langfuse_.py 
30
31
32
33
34
def get_langfuse_langchain_config() -> RunnableConfig:
    config: RunnableConfig = {
        "callbacks": [langfuse_context.get_current_langchain_handler()]
    }
    return config

langfuse_client_utils

ProcessMarketTrace

Bases: BaseModel

timestamp instance-attribute 
timestamp: int
market instance-attribute 
market: OmenAgentMarket
answer instance-attribute 
answer: CategoricalProbabilisticAnswer
trades instance-attribute 
trades: list[PlacedTrade]
timestamp_datetime property 
timestamp_datetime: DatetimeUTC
buy_trade property 
buy_trade: PlacedTrade | None
from_langfuse_trace staticmethod 
from_langfuse_trace(
    trace: TraceWithDetails,
) -> t.Optional[ProcessMarketTrace]
Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@staticmethod
def from_langfuse_trace(
    trace: TraceWithDetails,
) -> t.Optional["ProcessMarketTrace"]:
    market = trace_to_omen_agent_market(trace)
    answer = trace_to_answer(trace)
    trades = trace_to_trades(trace)

    if not market or not answer or not trades:
        return None

    return ProcessMarketTrace(
        market=market,
        answer=answer,
        trades=trades,
        timestamp=int(trace.timestamp.timestamp()),
    )

ResolvedBetWithTrace

Bases: BaseModel

bet instance-attribute 
bet: ResolvedBet
trace instance-attribute 
trace: ProcessMarketTrace

get_traces_for_agent 

get_traces_for_agent(
    agent_name: str,
    trace_name: str,
    from_timestamp: DatetimeUTC,
    has_output: bool,
    client: Langfuse,
    to_timestamp: DatetimeUTC | None = None,
    tags: str | list[str] | None = None,
) -> list[TraceWithDetails]

Fetch agent traces using pagination

Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
 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
def get_traces_for_agent(
    agent_name: str,
    trace_name: str,
    from_timestamp: DatetimeUTC,
    has_output: bool,
    client: Langfuse,
    to_timestamp: DatetimeUTC | None = None,
    tags: str | list[str] | None = None,
) -> list[TraceWithDetails]:
    """
    Fetch agent traces using pagination
    """
    total_pages = -1
    page = 1  # index starts from 1
    all_agent_traces = []
    while True:
        logger.debug(f"Fetching Langfuse page {page} / {total_pages}.")
        traces = client.fetch_traces(
            name=trace_name,
            limit=100,
            page=page,
            from_timestamp=from_timestamp,
            to_timestamp=to_timestamp,
            tags=tags,
        )
        if not traces.data:
            break
        page += 1
        total_pages = traces.meta.total_pages

        agent_traces = [
            t
            for t in traces.data
            if t.session_id is not None and agent_name in t.session_id
        ]
        if has_output:
            agent_traces = [t for t in agent_traces if t.output is not None]
        all_agent_traces.extend(agent_traces)
    return all_agent_traces

trace_to_omen_agent_market 

trace_to_omen_agent_market(
    trace: TraceWithDetails,
) -> OmenAgentMarket | None
Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def trace_to_omen_agent_market(trace: TraceWithDetails) -> OmenAgentMarket | None:
    if not trace.input:
        logger.warning(f"No input in the trace: {trace}")
        return None
    if not trace.input["args"]:
        logger.warning(f"No args in the trace: {trace}")
        return None
    assert len(trace.input["args"]) == 2 and trace.input["args"][0] == "omen"
    try:
        # If the market model is invalid (e.g. outdated), it will raise an exception
        market = OmenAgentMarket.model_validate(trace.input["args"][1])
        return market
    except Exception as e:
        logger.warning(f"Market not parsed from langfuse because: {e}")
        return None

trace_to_answer 

trace_to_answer(
    trace: TraceWithDetails,
) -> CategoricalProbabilisticAnswer
Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
121
122
123
124
def trace_to_answer(trace: TraceWithDetails) -> CategoricalProbabilisticAnswer:
    assert trace.output is not None, "Trace output is None"
    assert trace.output["answer"] is not None, "Trace output result is None"
    return CategoricalProbabilisticAnswer.model_validate(trace.output["answer"])

trace_to_trades 

trace_to_trades(
    trace: TraceWithDetails,
) -> list[PlacedTrade]
Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
127
128
129
130
def trace_to_trades(trace: TraceWithDetails) -> list[PlacedTrade]:
    assert trace.output is not None, "Trace output is None"
    assert trace.output["trades"] is not None, "Trace output trades is None"
    return [PlacedTrade.model_validate(t) for t in trace.output["trades"]]

get_closest_datetime_from_list 

get_closest_datetime_from_list(
    ref_datetime: DatetimeUTC, datetimes: list[DatetimeUTC]
) -> int

Get the index of the closest datetime to the reference datetime

Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
133
134
135
136
137
138
139
140
141
def get_closest_datetime_from_list(
    ref_datetime: DatetimeUTC, datetimes: list[DatetimeUTC]
) -> int:
    """Get the index of the closest datetime to the reference datetime"""
    if len(datetimes) == 1:
        return 0

    closest_datetime = min(datetimes, key=lambda dt: abs(dt - ref_datetime))
    return datetimes.index(closest_datetime)

get_trace_for_bet 

get_trace_for_bet(
    bet: ResolvedBet, traces: list[ProcessMarketTrace]
) -> ProcessMarketTrace | None
Source code in prediction_market_agent_tooling/tools/langfuse_client_utils.py 
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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def get_trace_for_bet(
    bet: ResolvedBet, traces: list[ProcessMarketTrace]
) -> ProcessMarketTrace | None:
    # Filter for traces with the same market id
    traces = [t for t in traces if t.market.id == bet.market_id]

    # Filter for traces with the same bet outcome and amount
    traces_for_bet: list[ProcessMarketTrace] = []
    for t in traces:
        if (
            t.market.collateral_token_contract_address_checksummed
            not in WRAPPED_XDAI_CONTRACT_ADDRESS
        ):
            # TODO: We need to compute bet amount token in USD here, but at the time of bet placement!
            logger.warning(
                "This currently works only for WXDAI markets, because we need to compare against USD value."
            )
            continue
        # Cannot use exact comparison due to gas fees
        if (
            t.buy_trade
            and t.buy_trade.outcome == bet.outcome
            and np.isclose(t.buy_trade.amount.value, bet.amount.value)
        ):
            traces_for_bet.append(t)

    if not traces_for_bet:
        return None
    elif len(traces_for_bet) == 1:
        return traces_for_bet[0]
    else:
        # In-case there are multiple traces for the same market, get the closest
        # trace to the bet
        closest_trace_index = get_closest_datetime_from_list(
            bet.created_time,
            [t.timestamp_datetime for t in traces_for_bet],
        )

        # Sanity check: Let's say the upper bound for time between
        # `agent.process_market` being called and the bet being placed is 20
        # minutes
        candidate_trace = traces_for_bet[closest_trace_index]
        if (
            abs(candidate_trace.timestamp_datetime - bet.created_time).total_seconds()
            > 1200
        ):
            logger.info(
                f"Closest trace to bet has timestamp {candidate_trace.timestamp}, "
                f"but bet was created at {bet.created_time}. Not matching"
            )
            return None

        return traces_for_bet[closest_trace_index]

omen

reality_accuracy

RealityAccuracyReport

Bases: BaseModel

total instance-attribute 
total: int
correct instance-attribute 
correct: int
accuracy property 
accuracy: float
reality_accuracy 
reality_accuracy(
    user: ChecksumAddress, since: timedelta
) -> RealityAccuracyReport
Source code in prediction_market_agent_tooling/tools/omen/reality_accuracy.py 
22
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
def reality_accuracy(user: ChecksumAddress, since: timedelta) -> RealityAccuracyReport:
    now = utcnow()
    start_from = now - since

    # Get all question ids where we placed the higher bond.
    user_responses = OmenSubgraphHandler().get_responses(
        limit=None,
        user=user,
        question_finalized_before=now,
        question_finalized_after=start_from,
    )
    unique_question_ids = set(r.question.questionId for r in user_responses)

    # Get all responses for these questions (including not ours)
    question_to_responses = {
        question_id: OmenSubgraphHandler().get_responses(
            limit=None, question_id=question_id
        )
        for question_id in unique_question_ids
    }

    total = 0
    correct = 0

    for question_id, responses in question_to_responses.items():
        is_correct = user_was_correct(user, responses)
        assert (
            is_correct is not None
        ), f"All these questions should be challenged by provided user: {responses[0].question.url}"

        total += 1
        correct += int(is_correct)

    return RealityAccuracyReport(total=total, correct=correct)
user_was_correct 
user_was_correct(
    user: ChecksumAddress, responses: list[RealityResponse]
) -> bool | None
Source code in prediction_market_agent_tooling/tools/omen/reality_accuracy.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def user_was_correct(
    user: ChecksumAddress, responses: list[RealityResponse]
) -> bool | None:
    sorted_responses = sorted(responses, key=lambda r: r.timestamp)
    users_sorted_responses = [r for r in sorted_responses if r.user_checksummed == user]

    if not users_sorted_responses:
        return None

    # Find the user's last response
    users_last_response = users_sorted_responses[-1]

    # Last response is the final one (if market is finalized)
    actual_resolution = sorted_responses[-1]

    # Compare the user's last answer with the actual resolution
    return users_last_response.answer == actual_resolution.answer

sell_positions

sell_all 
sell_all(
    api_keys: APIKeys,
    closing_later_than_days: int,
    auto_withdraw: bool = True,
) -> None

Helper function to sell all existing outcomes on Omen that would resolve later than in X days.

Source code in prediction_market_agent_tooling/tools/omen/sell_positions.py 
14
15
16
17
18
19
20
21
22
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
def sell_all(
    api_keys: APIKeys,
    closing_later_than_days: int,
    auto_withdraw: bool = True,
) -> None:
    """
    Helper function to sell all existing outcomes on Omen that would resolve later than in X days.
    """
    better_address = api_keys.bet_from_address
    bets = OmenSubgraphHandler().get_bets(
        better_address=better_address,
        market_opening_after=utcnow() + timedelta(days=closing_later_than_days),
    )
    bets_total_usd = sum(
        (b.get_collateral_amount_usd() for b in bets), start=USD.zero()
    )
    unique_market_urls = set(b.fpmm.url for b in bets)
    starting_balance = get_balances(better_address)
    new_balance = starting_balance

    logger.info(
        f"For {better_address}, found the following {len(bets)} bets on {len(unique_market_urls)} unique markets worth of {bets_total_usd} USD: {unique_market_urls}"
    )

    for bet in bets:
        agent_market = OmenAgentMarket.from_data_model(bet.fpmm)
        outcome = agent_market.outcomes[bet.outcomeIndex]
        current_token_balance = agent_market.get_token_balance(better_address, outcome)

        if current_token_balance.as_token <= agent_market.get_tiny_bet_amount():
            logger.info(
                f"Skipping bet on {bet.fpmm.url} because the actual balance is unreasonably low {current_token_balance}."
            )
            continue

        old_balance = new_balance
        bet_outcome = bet.fpmm.outcomes[bet.outcomeIndex]
        agent_market.sell_tokens(
            outcome=bet_outcome,
            amount=current_token_balance,
            auto_withdraw=auto_withdraw,
            api_keys=api_keys,
        )
        new_balance = get_balances(better_address)

        logger.info(
            f"Sold bet on {bet.fpmm.url} for {new_balance.wxdai - old_balance.wxdai} xDai."
        )

    logger.info(f"Obtained back {new_balance.wxdai - starting_balance.wxdai} wxDai.")

parallelism

A module-attribute 

A = TypeVar('A')

B module-attribute 

B = TypeVar('B')

par_map 

par_map(
    items: list[A],
    func: Callable[[A], B],
    max_workers: int = 5,
) -> list[B]

Applies the function to each element using the specified executor. Awaits for all results.

Source code in prediction_market_agent_tooling/tools/parallelism.py 
11
12
13
14
15
16
17
18
19
20
21
22
def par_map(
    items: list[A],
    func: Callable[[A], B],
    max_workers: int = 5,
) -> "list[B]":
    """Applies the function to each element using the specified executor. Awaits for all results."""
    executor = get_reusable_executor(max_workers=max_workers, initializer=patch_logger)
    futures = [executor.submit(func, item) for item in items]
    results = []
    for fut in futures:
        results.append(fut.result())
    return results

par_generator 

par_generator(
    items: list[A],
    func: Callable[[A], B],
    max_workers: int = 5,
) -> Generator[B, None, None]

Applies the function to each element using the specified executor. Yields results as they come.

Source code in prediction_market_agent_tooling/tools/parallelism.py 
25
26
27
28
29
30
31
32
33
def par_generator(
    items: list[A],
    func: Callable[[A], B],
    max_workers: int = 5,
) -> Generator[B, None, None]:
    """Applies the function to each element using the specified executor. Yields results as they come."""
    executor = get_reusable_executor(max_workers=max_workers, initializer=patch_logger)
    for res in executor.map(func, items):
        yield res

relevant_news_analysis

data_models

RelevantNewsAnalysis

Bases: BaseModel

reasoning class-attribute instance-attribute 
reasoning: str = Field(
    ...,
    description="The reason why the news contains information relevant to the given question. Or if no news is relevant, why not.",
)
contains_relevant_news class-attribute instance-attribute 
contains_relevant_news: bool = Field(
    ...,
    description="A boolean flag for whether the news contains information relevant to the given question.",
)
RelevantNews

Bases: BaseModel

question instance-attribute 
question: str
url instance-attribute 
url: str
summary instance-attribute 
summary: str
relevance_reasoning instance-attribute 
relevance_reasoning: str
days_ago instance-attribute 
days_ago: int
from_tavily_result_and_analysis staticmethod 
from_tavily_result_and_analysis(
    question: str,
    days_ago: int,
    tavily_result: TavilyResult,
    relevant_news_analysis: RelevantNewsAnalysis,
) -> RelevantNews
Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/data_models.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@staticmethod
def from_tavily_result_and_analysis(
    question: str,
    days_ago: int,
    tavily_result: TavilyResult,
    relevant_news_analysis: RelevantNewsAnalysis,
) -> "RelevantNews":
    return RelevantNews(
        question=question,
        url=tavily_result.url,
        summary=tavily_result.content,
        relevance_reasoning=relevant_news_analysis.reasoning,
        days_ago=days_ago,
    )
NoRelevantNews

Bases: BaseModel

A placeholder model for when no relevant news is found. Enables ability to distinguish between 'a cache hit with no news' and 'a cache miss'.

relevant_news_analysis

SUMMARISE_RELEVANT_NEWS_PROMPT_TEMPLATE module-attribute 
SUMMARISE_RELEVANT_NEWS_PROMPT_TEMPLATE = "\nYou are an expert news analyst, tracking stories that may affect your prediction to the outcome of a particular QUESTION.\n\nYour role is to identify only the relevant information from a scraped news site (RAW_CONTENT), analyse it, and determine whether it contains developments or announcements occurring **after** the DATE_OF_INTEREST that could affect the outcome of the QUESTION.\n\nNote that the news article may be published after the DATE_OF_INTEREST, but reference information that is older than the DATE_OF_INTEREST.\n\n[QUESTION]\n{question}\n\n[DATE_OF_INTEREST]\n{date_of_interest}\n\n[RAW_CONTENT]\n{raw_content}\n\nFor your analysis, you should:\n- Discard the 'noise' from the raw content (e.g. ads, irrelevant content)\n- Consider ONLY information that would have a notable impact on the outcome of the question.\n- Consider ONLY information relating to an announcement or development that occurred **after** the DATE_OF_INTEREST.\n- Present this information concisely in your reasoning.\n- In your reasoning, do not use the term 'DATE_OF_INTEREST' directly. Use the actual date you are referring to instead.\n- In your reasoning, do not use the term 'RAW_CONTENT' directly. Refer to it as 'the article', or quote the content you are referring to.\n\n{format_instructions}\n"
analyse_news_relevance 
analyse_news_relevance(
    raw_content: str,
    question: str,
    date_of_interest: date,
    model: str,
    temperature: float,
) -> RelevantNewsAnalysis

Analyse whether the news contains new (relative to the given date) information relevant to the given question.

Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/relevant_news_analysis.py 
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
@observe()
def analyse_news_relevance(
    raw_content: str,
    question: str,
    date_of_interest: date,
    model: str,
    temperature: float,
) -> RelevantNewsAnalysis:
    """
    Analyse whether the news contains new (relative to the given date)
    information relevant to the given question.
    """
    parser = PydanticOutputParser(pydantic_object=RelevantNewsAnalysis)
    prompt = PromptTemplate(
        template=SUMMARISE_RELEVANT_NEWS_PROMPT_TEMPLATE,
        input_variables=["question", "date_of_interest", "raw_content"],
        partial_variables={"format_instructions": parser.get_format_instructions()},
    )
    llm = ChatOpenAI(
        temperature=temperature,
        model_name=model,
        openai_api_key=APIKeys().openai_api_key,
    )
    chain = prompt | llm | parser

    relevant_news_analysis: RelevantNewsAnalysis = chain.invoke(
        {
            "raw_content": raw_content,
            "question": question,
            "date_of_interest": str(date_of_interest),
        },
        config=get_langfuse_langchain_config(),
    )
    return relevant_news_analysis
get_certified_relevant_news_since 
get_certified_relevant_news_since(
    question: str, days_ago: int
) -> RelevantNews | None

Get relevant news since a given date for a given question. Retrieves possibly relevant news from tavily, then checks that it is relevant via an LLM call.

Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/relevant_news_analysis.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@observe()
def get_certified_relevant_news_since(
    question: str,
    days_ago: int,
) -> RelevantNews | None:
    """
    Get relevant news since a given date for a given question. Retrieves
    possibly relevant news from tavily, then checks that it is relevant via
    an LLM call.
    """
    news_since = date.today() - timedelta(days=days_ago)
    results = get_relevant_news_since(
        question=question,
        news_since=news_since,
        score_threshold=0.0,  # Be conservative to avoid missing relevant information
        max_results=3,  # A tradeoff between cost and quality. 3 seems to be a good balance.
    )

    # Sort results by descending 'relevance score' to maximise the chance of
    # finding relevant news early
    results = sorted(
        results,
        key=lambda result: result.score,
        reverse=True,
    )

    for result in results:
        relevant_news_analysis = analyse_news_relevance(
            raw_content=check_not_none(result.raw_content),
            question=question,
            date_of_interest=news_since,
            model="gpt-4o",  # 4o-mini isn't good enough, 1o and 1o-mini are too expensive
            temperature=0.0,
        )

        # Return first relevant news found
        if relevant_news_analysis.contains_relevant_news:
            return RelevantNews.from_tavily_result_and_analysis(
                question=question,
                days_ago=days_ago,
                tavily_result=result,
                relevant_news_analysis=relevant_news_analysis,
            )

    # No relevant news found
    return None
get_certified_relevant_news_since_cached 
get_certified_relevant_news_since_cached(
    question: str,
    days_ago: int,
    cache: RelevantNewsResponseCache,
) -> RelevantNews | None
Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/relevant_news_analysis.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def get_certified_relevant_news_since_cached(
    question: str,
    days_ago: int,
    cache: RelevantNewsResponseCache,
) -> RelevantNews | None:
    cached = cache.find(question=question, days_ago=days_ago)

    if isinstance(cached, NoRelevantNews):
        return None
    elif cached is None:
        relevant_news = get_certified_relevant_news_since(
            question=question,
            days_ago=days_ago,
        )
        cache.save(
            question=question,
            days_ago=days_ago,
            relevant_news=relevant_news,
        )
        return relevant_news
    else:
        return cached

relevant_news_cache

RelevantNewsCacheModel

Bases: SQLModel

id class-attribute instance-attribute 
id: int | None = Field(default=None, primary_key=True)
question class-attribute instance-attribute 
question: str = Field(index=True)
datetime_ class-attribute instance-attribute 
datetime_: datetime = Field(index=True)
days_ago instance-attribute 
days_ago: int
json_dump instance-attribute 
json_dump: str | None
RelevantNewsResponseCache 
RelevantNewsResponseCache(api_keys: APIKeys | None = None)
Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/relevant_news_cache.py 
27
28
29
30
31
def __init__(self, api_keys: APIKeys | None = None):
    self.db_manager = DBManager(
        (api_keys or APIKeys()).sqlalchemy_db_url.get_secret_value()
    )
    self._initialize_db()
db_manager instance-attribute 
db_manager = DBManager(get_secret_value())
find 
find(
    question: str, days_ago: int
) -> RelevantNews | NoRelevantNews | None
Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/relevant_news_cache.py 
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
def find(
    self,
    question: str,
    days_ago: int,
) -> RelevantNews | NoRelevantNews | None:
    with self.db_manager.get_session() as session:
        query = (
            select(RelevantNewsCacheModel)
            .where(RelevantNewsCacheModel.question == question)
            .where(RelevantNewsCacheModel.days_ago <= days_ago)
            .where(
                RelevantNewsCacheModel.datetime_ >= utcnow() - timedelta(days=1)
            )  # Cache entries expire after 1 day
        )
        item = session.exec(
            query.order_by(desc(RelevantNewsCacheModel.datetime_))
        ).first()

        if item is None:
            return None
        else:
            if item.json_dump is None:
                return NoRelevantNews()
            else:
                try:
                    return RelevantNews.model_validate_json(item.json_dump)
                except ValidationError as e:
                    logger.error(
                        f"Error deserializing RelevantNews from cache for {question=}, {days_ago=} and {item=}: {e}"
                    )
                    return None
save 
save(
    question: str,
    days_ago: int,
    relevant_news: RelevantNews | None,
) -> None
Source code in prediction_market_agent_tooling/tools/relevant_news_analysis/relevant_news_cache.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def save(
    self,
    question: str,
    days_ago: int,
    relevant_news: RelevantNews | None,
) -> None:
    with self.db_manager.get_session() as session:
        cached = RelevantNewsCacheModel(
            question=question,
            days_ago=days_ago,
            datetime_=utcnow(),  # Assumes that the cache is being updated at the time the news is found
            json_dump=relevant_news.model_dump_json() if relevant_news else None,
        )
        session.add(cached)
        session.commit()

safe

create_safe 

create_safe(
    ethereum_client: EthereumClient,
    account: LocalAccount,
    owners: list[str],
    salt_nonce: int,
    threshold: int = 1,
    without_events: bool = False,
) -> ChecksumAddress | None
Source code in prediction_market_agent_tooling/tools/safe.py 
 20
 21
 22
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def create_safe(
    ethereum_client: EthereumClient,
    account: LocalAccount,
    owners: list[str],
    salt_nonce: int,
    threshold: int = 1,
    without_events: bool = False,  # following safe-cli convention
) -> ChecksumAddress | None:
    to = NULL_ADDRESS
    data = b""
    payment_token = NULL_ADDRESS
    payment = 0
    payment_receiver = NULL_ADDRESS

    if len(owners) < threshold:
        raise ValueError("Threshold cannot be bigger than the number of unique owners")

    ethereum_network = ethereum_client.get_network()

    safe_contract_address = (
        get_safe_contract_address(ethereum_client)
        if without_events
        else get_safe_l2_contract_address(ethereum_client)
    )
    proxy_factory_address = get_proxy_factory_address(ethereum_client)
    fallback_handler = get_default_fallback_handler_address(ethereum_client)

    if not ethereum_client.is_contract(safe_contract_address):
        raise EnvironmentError(
            f"Safe contract address {safe_contract_address} "
            f"does not exist on network {ethereum_network.name}"
        )
    elif not ethereum_client.is_contract(proxy_factory_address):
        raise EnvironmentError(
            f"Proxy contract address {proxy_factory_address} "
            f"does not exist on network {ethereum_network.name}"
        )
    elif fallback_handler != NULL_ADDRESS and not ethereum_client.is_contract(
        fallback_handler
    ):
        raise EnvironmentError(
            f"Fallback handler address {fallback_handler} "
            f"does not exist on network {ethereum_network.name}"
        )

    account_balance = Wei(ethereum_client.get_balance(account.address))
    account_balance_xdai = account_balance.as_token
    # We set a reasonable expected balance below for Safe deployment not to fail.
    if account_balance_xdai.value < 0.01:
        raise ValueError(
            f"Client's balance is {account_balance_xdai} xDAI, too low for deploying a Safe."
        )

    logger.info(
        f"Network {ethereum_client.get_network().name} - Sender {account.address} - "
        f"Balance: {account_balance_xdai} xDAI"
    )

    if not ethereum_client.w3.eth.get_code(
        safe_contract_address
    ) or not ethereum_client.w3.eth.get_code(proxy_factory_address):
        raise EnvironmentError("Network not supported")

    logger.info(
        f"Creating new Safe with owners={owners} threshold={threshold} salt-nonce={salt_nonce}"
    )

    # We ignore mypy below because using the proper class SafeV141 yields an error and mypy
    # doesn't understand that there is a hacky factory method (__new__) on this abstract class.
    safe_version = SafeV141(safe_contract_address, ethereum_client).retrieve_version()
    logger.info(
        f"Safe-master-copy={safe_contract_address} version={safe_version}\n"
        f"Fallback-handler={fallback_handler}\n"
        f"Proxy factory={proxy_factory_address}"
    )

    # Note that by default a safe contract instance of version 1.4.1 will be fetched (determined
    # by safe_contract_address), but if a different address (corresponding to an older version, e.g. 1.3.0)
    # is passed, it will also work, since older versions also have the setup method.
    safe_contract = get_safe_V1_4_1_contract(ethereum_client.w3, safe_contract_address)
    safe_creation_tx_data = HexBytes(
        safe_contract.functions.setup(
            owners,
            threshold,
            to,
            data,
            fallback_handler,
            payment_token,
            payment,
            payment_receiver,
        ).build_transaction({"gas": 1, "gasPrice": Wei(1).value})["data"]
    )

    proxy_factory = ProxyFactoryV141(proxy_factory_address, ethereum_client)
    expected_safe_address = proxy_factory.calculate_proxy_address(
        safe_contract_address, safe_creation_tx_data, salt_nonce
    )
    if ethereum_client.is_contract(expected_safe_address):
        logger.info(f"Safe on {expected_safe_address} is already deployed")
        return expected_safe_address

    ethereum_tx_sent = proxy_factory.deploy_proxy_contract_with_nonce(
        account, safe_contract_address, safe_creation_tx_data, salt_nonce
    )
    logger.info(
        f"Sent tx with tx-hash={ethereum_tx_sent.tx_hash.hex()} "
        f"Safe={ethereum_tx_sent.contract_address} is being created"
    )
    logger.info(f"Tx parameters={ethereum_tx_sent.tx}")
    return ethereum_tx_sent.contract_address

singleton

SingletonMeta

Bases: type, Generic[_T]

The Singleton class can be implemented in different ways in Python. Some possible methods include: base class, decorator, metaclass. We will use the metaclass because it is best suited for this purpose.

streamlit_user_login

LoggedInUser

Bases: BaseModel

email instance-attribute 
email: str
password instance-attribute 
password: SecretStr

LoginSettings

Bases: BaseSettings

model_config class-attribute instance-attribute 
model_config = SettingsConfigDict(
    env_file=".env",
    env_file_encoding="utf-8",
    extra="ignore",
)
free_for_everyone class-attribute instance-attribute 
free_for_everyone: bool = False
free_access_codes class-attribute instance-attribute 
free_access_codes: list[SecretStr] = []
users class-attribute instance-attribute 
users: list[LoggedInUser] = []

LoggedEnum

Bases: str, Enum

SELF_PAYING class-attribute instance-attribute 
SELF_PAYING = 'self_paying'
FREE_ACCESS class-attribute instance-attribute 
FREE_ACCESS = 'free_access'
USER_LOGGED_IN class-attribute instance-attribute 
USER_LOGGED_IN = 'user_logged_in'

find_logged_in_user 

find_logged_in_user(
    email: str, password: SecretStr
) -> LoggedInUser | None
Source code in prediction_market_agent_tooling/tools/streamlit_user_login.py 
30
31
32
33
34
35
36
37
38
39
40
def find_logged_in_user(email: str, password: SecretStr) -> LoggedInUser | None:
    login_settings = LoginSettings()

    for user in login_settings.users:
        if (
            user.email == email
            and user.password.get_secret_value() == password.get_secret_value()
        ):
            return user

    return None

streamlit_login 

streamlit_login() -> tuple[LoggedEnum, LoggedInUser | None]
Source code in prediction_market_agent_tooling/tools/streamlit_user_login.py 
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
def streamlit_login() -> tuple[LoggedEnum, LoggedInUser | None]:
    login_settings = LoginSettings()

    if login_settings.free_for_everyone:
        logger.info("Free access for everyone!")
        return LoggedEnum.FREE_ACCESS, None

    if (
        free_access_code := st.query_params.get("free_access_code")
    ) is not None and free_access_code in [
        x.get_secret_value() for x in login_settings.free_access_codes
    ]:
        logger.info(f"Using free access code: {free_access_code}.")
        return LoggedEnum.FREE_ACCESS, None

    # TODO: Because we are initializing APIKeys (based on environment variables) in random places in the code, the user can not provide their own access keys, such as OPENAI_API_KEY.
    # Doing that would require to change the environment variables in the code, but Streamlit is using 1 thread per session, and the environment variables are shared between threads.
    # So other users would get the secrets provided by the first user!
    # Another option is to refactor the code, such that APIKeys are initialized in a single place, somewhere in the agent's initialisation and then, we can initialize them here based on the user's input and just provide them  to the agent.
    # After that's done, we should also have unique api keys per registered user.
    self_paying = False
    # self_paying = st.checkbox("I will provide my own access keys", value=False)

    if not self_paying:
        email = st.text_input("Email")
        password = SecretStr(st.text_input("Password", type="password"))
        logged_user = find_logged_in_user(email, password)

        if logged_user is not None:
            logger.info(f"Logged in as {email}.")
            return LoggedEnum.USER_LOGGED_IN, logged_user

        else:
            st.error("Invalid email or password.")
            st.stop()

    else:
        raise NotImplementedError(
            "Should not happen. Self-paying is not implemented yet. See the comment above."
        )

tavily

tavily_models

TavilyResult

Bases: BaseModel

title instance-attribute 
title: str
url instance-attribute 
url: str
content instance-attribute 
content: str
score instance-attribute 
score: float
raw_content instance-attribute 
raw_content: str | None
TavilyResponse

Bases: BaseModel

query instance-attribute 
query: str
follow_up_questions class-attribute instance-attribute 
follow_up_questions: str | None = None
answer instance-attribute 
answer: str
images instance-attribute 
images: list[str]
results instance-attribute 
results: list[TavilyResult]
response_time instance-attribute 
response_time: float
DEFAULT_SCORE_THRESHOLD module-attribute 
DEFAULT_SCORE_THRESHOLD = 0.75

tavily_search(
    query: str,
    search_depth: Literal["basic", "advanced"] = "advanced",
    topic: Literal["general", "news"] = "general",
    news_since: date | None = None,
    max_results: int = 5,
    include_domains: Sequence[str] | None = None,
    exclude_domains: Sequence[str] | None = None,
    include_answer: bool = True,
    include_raw_content: bool = True,
    include_images: bool = True,
    use_cache: bool = False,
    api_keys: APIKeys | None = None,
) -> TavilyResponse

Argument default values are different from the original method, to return everything by default, because it can be handy in the future and it doesn't increase the costs.

Source code in prediction_market_agent_tooling/tools/tavily/tavily_search.py 
17
18
19
20
21
22
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
@db_cache(
    max_age=timedelta(days=1),
    ignore_args=["api_keys"],
    log_error_on_unsavable_data=False,
)
def tavily_search(
    query: str,
    search_depth: t.Literal["basic", "advanced"] = "advanced",
    topic: t.Literal["general", "news"] = "general",
    news_since: date | None = None,
    max_results: int = 5,
    include_domains: t.Sequence[str] | None = None,
    exclude_domains: t.Sequence[str] | None = None,
    include_answer: bool = True,
    include_raw_content: bool = True,
    include_images: bool = True,
    use_cache: bool = False,
    api_keys: APIKeys | None = None,
) -> TavilyResponse:
    """
    Argument default values are different from the original method, to return everything by default, because it can be handy in the future and it doesn't increase the costs.
    """
    if topic == "news" and news_since is None:
        raise ValueError("When topic is 'news', news_since must be provided")
    if topic == "general" and news_since is not None:
        raise ValueError("When topic is 'general', news_since must be None")

    days = None if news_since is None else (date.today() - news_since).days
    response = _tavily_search(
        query=query,
        search_depth=search_depth,
        topic=topic,
        max_results=max_results,
        days=days,
        include_domains=include_domains,
        exclude_domains=exclude_domains,
        include_answer=include_answer,
        include_raw_content=include_raw_content,
        include_images=include_images,
        use_cache=use_cache,
        api_keys=api_keys,
    )
    response_parsed = TavilyResponse.model_validate(response)

    return response_parsed
get_relevant_news_since 
get_relevant_news_since(
    question: str,
    news_since: date,
    score_threshold: float = DEFAULT_SCORE_THRESHOLD,
    max_results: int = 3,
) -> list[TavilyResult]
Source code in prediction_market_agent_tooling/tools/tavily/tavily_search.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
def get_relevant_news_since(
    question: str,
    news_since: date,
    score_threshold: float = DEFAULT_SCORE_THRESHOLD,
    max_results: int = 3,
) -> list[TavilyResult]:
    news = tavily_search(
        query=question,
        news_since=news_since,
        max_results=max_results,
        topic="news",
    )
    return [r for r in news.results if r.score > score_threshold]

transaction_cache

TransactionBlockCache 

TransactionBlockCache(web3: Web3)
Source code in prediction_market_agent_tooling/tools/transaction_cache.py 
11
12
13
14
def __init__(self, web3: Web3):
    self.block_number_cache = dc.Cache(".cache/block_cache_dir")
    self.block_timestamp_cache = dc.Cache(".cache/timestamp_cache_dir")
    self.web3 = web3
block_number_cache instance-attribute 
block_number_cache = Cache('.cache/block_cache_dir')
block_timestamp_cache instance-attribute 
block_timestamp_cache = Cache('.cache/timestamp_cache_dir')
web3 instance-attribute 
web3 = web3
fetch_block_number 
fetch_block_number(transaction_hash: str) -> int
Source code in prediction_market_agent_tooling/tools/transaction_cache.py 
16
17
18
19
20
21
22
23
@tenacity.retry(
    wait=wait_exponential(multiplier=1, min=1, max=4),
    stop=tenacity.stop_after_attempt(7),
    after=lambda x: logger.debug(f"fetch tx failed, {x.attempt_number=}."),
)
def fetch_block_number(self, transaction_hash: str) -> int:
    tx = self.web3.eth.get_transaction(HexStr(transaction_hash))
    return tx["blockNumber"]
fetch_block_timestamp 
fetch_block_timestamp(block_number: int) -> int
Source code in prediction_market_agent_tooling/tools/transaction_cache.py 
25
26
27
28
29
30
31
32
@tenacity.retry(
    wait=wait_exponential(multiplier=1, min=1, max=4),
    stop=tenacity.stop_after_attempt(7),
    after=lambda x: logger.debug(f"fetch tx failed, {x.attempt_number=}."),
)
def fetch_block_timestamp(self, block_number: int) -> int:
    block = self.web3.eth.get_block(block_number)
    return block["timestamp"]
get_block_number 
get_block_number(tx_hash: str) -> int
Source code in prediction_market_agent_tooling/tools/transaction_cache.py 
34
35
36
37
38
39
40
def get_block_number(self, tx_hash: str) -> int:
    if tx_hash in self.block_number_cache:
        return int(self.block_number_cache[tx_hash])

    block_number = self.fetch_block_number(tx_hash)
    self.block_number_cache[tx_hash] = block_number
    return block_number
get_block_timestamp 
get_block_timestamp(block_number: int) -> int
Source code in prediction_market_agent_tooling/tools/transaction_cache.py 
42
43
44
45
46
47
48
def get_block_timestamp(self, block_number: int) -> int:
    if block_number in self.block_timestamp_cache:
        return int(self.block_timestamp_cache[block_number])

    block_timestamp = self.fetch_block_timestamp(block_number)
    self.block_timestamp_cache[block_number] = block_timestamp
    return block_timestamp

utils

T module-attribute 

T = TypeVar('T')

LLM_SUPER_LOW_TEMPERATURE module-attribute 

LLM_SUPER_LOW_TEMPERATURE = 1e-08

LLM_SEED module-attribute 

LLM_SEED = 0

BPS_CONSTANT module-attribute 

BPS_CONSTANT = 10000

BaseModelT module-attribute 

BaseModelT = TypeVar('BaseModelT', bound=BaseModel)

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 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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

should_not_happen 

should_not_happen(
    msg: str = "Should not happen.",
    exp: Type[ValueError] = ValueError,
) -> NoReturn

Utility function to raise an exception with a message.

Handy for cases like this:

return (
    1 if variable == X
    else 2 if variable == Y
    else 3 if variable == Z
    else should_not_happen(f"Variable {variable} is unknown.")
)

To prevent silent bugs with useful error message.

Source code in prediction_market_agent_tooling/tools/utils.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def should_not_happen(
    msg: str = "Should not happen.", exp: Type[ValueError] = ValueError
) -> NoReturn:
    """
    Utility function to raise an exception with a message.

    Handy for cases like this:

    ```
    return (
        1 if variable == X
        else 2 if variable == Y
        else 3 if variable == Z
        else should_not_happen(f"Variable {variable} is unknown.")
    )
    ```

    To prevent silent bugs with useful error message.
    """
    raise exp(msg)

export_requirements_from_toml 

export_requirements_from_toml(output_dir: str) -> None
Source code in prediction_market_agent_tooling/tools/utils.py 
80
81
82
83
84
85
86
87
88
89
def export_requirements_from_toml(output_dir: str) -> None:
    if not os.path.exists(output_dir):
        raise ValueError(f"Directory {output_dir} does not exist")
    output_file = f"{output_dir}/requirements.txt"
    subprocess.run(
        f"poetry export -f requirements.txt --without-hashes --output {output_file}",
        shell=True,
        check=True,
    )
    logger.debug(f"Saved requirements to {output_dir}/requirements.txt")

utcnow 

utcnow() -> DatetimeUTC
Source code in prediction_market_agent_tooling/tools/utils.py 
92
93
def utcnow() -> DatetimeUTC:
    return DatetimeUTC.from_datetime(datetime.now(pytz.UTC))

utc_datetime 

utc_datetime(
    year: int,
    month: int,
    day: int,
    hour: int = 0,
    minute: int = 0,
    second: int = 0,
    microsecond: int = 0,
    *,
    fold: int = 0,
) -> DatetimeUTC
Source code in prediction_market_agent_tooling/tools/utils.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def utc_datetime(
    year: int,
    month: int,
    day: int,
    hour: int = 0,
    minute: int = 0,
    second: int = 0,
    microsecond: int = 0,
    *,
    fold: int = 0,
) -> DatetimeUTC:
    dt = datetime(
        year=year,
        month=month,
        day=day,
        hour=hour,
        minute=minute,
        second=second,
        microsecond=microsecond,
        tzinfo=pytz.UTC,
        fold=fold,
    )
    return DatetimeUTC.from_datetime(dt)

get_current_git_commit_sha 

get_current_git_commit_sha() -> str
Source code in prediction_market_agent_tooling/tools/utils.py 
121
122
123
124
125
def get_current_git_commit_sha() -> str:
    # Import here to avoid erroring out if git repository is not present, but the function is not used anyway.
    import git

    return git.Repo(search_parent_directories=True).head.commit.hexsha

to_int_timestamp 

to_int_timestamp(dt: datetime) -> int
Source code in prediction_market_agent_tooling/tools/utils.py 
128
129
def to_int_timestamp(dt: datetime) -> int:
    return int(dt.timestamp())

get_current_git_branch 

get_current_git_branch() -> str
Source code in prediction_market_agent_tooling/tools/utils.py 
132
133
134
135
136
def get_current_git_branch() -> str:
    # Import here to avoid erroring out if git repository is not present, but the function is not used anyway.
    import git

    return git.Repo(search_parent_directories=True).active_branch.name

get_current_git_url 

get_current_git_url() -> str
Source code in prediction_market_agent_tooling/tools/utils.py 
139
140
141
142
143
def get_current_git_url() -> str:
    # Import here to avoid erroring out if git repository is not present, but the function is not used anyway.
    import git

    return git.Repo(search_parent_directories=True).remotes.origin.url

response_to_json 

response_to_json(response: Response) -> dict[str, Any]
Source code in prediction_market_agent_tooling/tools/utils.py 
146
147
148
149
def response_to_json(response: requests.models.Response) -> dict[str, Any]:
    response.raise_for_status()
    response_json: dict[str, Any] = response.json()
    return response_json

response_to_model 

response_to_model(
    response: Response, model: Type[BaseModelT]
) -> BaseModelT
Source code in prediction_market_agent_tooling/tools/utils.py 
155
156
157
158
159
160
161
162
def response_to_model(
    response: requests.models.Response, model: Type[BaseModelT]
) -> BaseModelT:
    response_json = response_to_json(response)
    try:
        return model.model_validate(response_json)
    except ValidationError as e:
        raise ValueError(f"Unable to validate: `{response_json}`") from e

response_list_to_model 

response_list_to_model(
    response: Response, model: Type[BaseModelT]
) -> list[BaseModelT]
Source code in prediction_market_agent_tooling/tools/utils.py 
165
166
167
168
169
170
171
172
def response_list_to_model(
    response: requests.models.Response, model: Type[BaseModelT]
) -> list[BaseModelT]:
    response_json = response_to_json(response)
    try:
        return [model.model_validate(x) for x in response_json]
    except ValidationError as e:
        raise ValueError(f"Unable to validate: `{response_json}`") from e

secret_str_from_env 

secret_str_from_env(key: str) -> SecretStr | None
Source code in prediction_market_agent_tooling/tools/utils.py 
175
176
177
def secret_str_from_env(key: str) -> SecretStr | None:
    value = os.getenv(key)
    return SecretStr(value) if value else None

prob_uncertainty 

prob_uncertainty(prob: Probability) -> float

Returns a value between 0 and 1, where 0 means the market is not uncertain at all and 1 means it's completely uncertain.

Examples:

  • Market's probability is 0.5, so the market is completely uncertain: prob_uncertainty(0.5) == 1
  • Market's probability is 0.1, so the market is quite certain about NO: prob_uncertainty(0.1) == 0.468
  • Market's probability is 0.95, so the market is quite certain about YES: prob_uncertainty(0.95) == 0.286
Source code in prediction_market_agent_tooling/tools/utils.py 
180
181
182
183
184
185
186
187
188
189
def prob_uncertainty(prob: Probability) -> float:
    """
    Returns a value between 0 and 1, where 0 means the market is not uncertain at all and 1 means it's completely uncertain.

    Examples:
        - Market's probability is 0.5, so the market is completely uncertain: prob_uncertainty(0.5) == 1
        - Market's probability is 0.1, so the market is quite certain about NO: prob_uncertainty(0.1) == 0.468
        - Market's probability is 0.95, so the market is quite certain about YES: prob_uncertainty(0.95) == 0.286
    """
    return float(entropy([prob, 1 - prob], base=2))

calculate_sell_amount_in_collateral 

calculate_sell_amount_in_collateral(
    shares_to_sell: OutcomeToken,
    outcome_index: int,
    pool_balances: list[OutcomeToken],
    fees: MarketFees,
) -> CollateralToken

Computes the amount of collateral that needs to be sold to get shares amount of shares. Returns None if the amount can't be computed.

Taken from https://github.com/protofire/omen-exchange/blob/29d0ab16bdafa5cc0d37933c1c7608a055400c73/app/src/util/tools/fpmm/trading/index.ts#L99 Simplified for binary markets.

Source code in prediction_market_agent_tooling/tools/utils.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def calculate_sell_amount_in_collateral(
    shares_to_sell: OutcomeToken,
    outcome_index: int,
    pool_balances: list[OutcomeToken],
    fees: MarketFees,
) -> CollateralToken:
    """
    Computes the amount of collateral that needs to be sold to get `shares`
    amount of shares. Returns None if the amount can't be computed.

    Taken from https://github.com/protofire/omen-exchange/blob/29d0ab16bdafa5cc0d37933c1c7608a055400c73/app/src/util/tools/fpmm/trading/index.ts#L99
    Simplified for binary markets.
    """

    if shares_to_sell == 0:
        return CollateralToken(0)

    if not (0 <= outcome_index < len(pool_balances)):
        raise IndexError("Invalid outcome index")

    if any([v <= 0 for v in pool_balances]):
        raise ValueError("All pool balances must be greater than 0")

    holdings = pool_balances[outcome_index]
    other_holdings = [v for i, v in enumerate(pool_balances) if i != outcome_index]

    def f(r: float) -> float:
        R = (r + fees.absolute) / (1 - fees.bet_proportion)

        # First term: product of (h_i - R) for i != outcome_index
        first_term = prod(h.value - R for h in other_holdings)

        # Second term: (h_o + s - R)
        second_term = holdings.value + shares_to_sell.value - R

        # Third term: product of all holdings (including outcome_index)
        total_product = prod(h.value for h in pool_balances)

        return (first_term * second_term) - total_product

    amount_to_sell = newton(f, 0)
    return CollateralToken(float(amount_to_sell) * 0.999999)  # Avoid rounding errors

extract_error_from_retry_error 

extract_error_from_retry_error(
    e: BaseException | RetryError,
) -> BaseException
Source code in prediction_market_agent_tooling/tools/utils.py 
236
237
238
239
240
241
242
def extract_error_from_retry_error(e: BaseException | RetryError) -> BaseException:
    if (
        isinstance(e, RetryError)
        and (exp_from_retry := e.last_attempt.exception()) is not None
    ):
        e = exp_from_retry
    return e

web3_utils

ONE_NONCE module-attribute 

ONE_NONCE = Nonce(1)

ONE_XDAI module-attribute 

ONE_XDAI = xDai(1)

ZERO_BYTES module-attribute 

ZERO_BYTES = HexBytes(HASH_ZERO)

NOT_REVERTED_ICASE_REGEX_PATTERN module-attribute 

NOT_REVERTED_ICASE_REGEX_PATTERN = '(?i)(?!.*reverted.*)'

generate_private_key 

generate_private_key() -> PrivateKey
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
39
40
def generate_private_key() -> PrivateKey:
    return private_key_type("0x" + secrets.token_hex(32))

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 
43
44
45
def private_key_to_public_key(private_key: SecretStr) -> ChecksumAddress:
    account = Account.from_key(private_key.get_secret_value())
    return verify_address(account.address)

verify_address 

verify_address(address: str) -> ChecksumAddress
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
48
49
50
51
52
53
def verify_address(address: str) -> ChecksumAddress:
    if not Web3.is_checksum_address(address):
        raise ValueError(
            f"The address {address} is not a valid checksum address, please fix your input."
        )
    return ChecksumAddress(HexAddress(HexStr(address)))

check_tx_receipt 

check_tx_receipt(receipt: TxReceipt) -> None
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
56
57
58
59
60
def check_tx_receipt(receipt: TxReceipt) -> None:
    if receipt["status"] != 1:
        raise ValueError(
            f"Transaction failed with status code {receipt['status']}. Receipt: {receipt}"
        )

unwrap_generic_value 

unwrap_generic_value(value: Any) -> Any
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
63
64
65
66
67
68
69
70
71
72
73
74
def unwrap_generic_value(value: Any) -> Any:
    if value is None:
        return None
    if isinstance(value, _GenericValue):
        return value.value
    elif isinstance(value, list):
        return [unwrap_generic_value(v) for v in value]
    elif isinstance(value, tuple):
        return tuple(unwrap_generic_value(v) for v in value)
    elif isinstance(value, dict):
        return {k: unwrap_generic_value(v) for k, v in value.items()}
    return value

parse_function_params 

parse_function_params(
    params: Optional[
        list[Any] | tuple[Any] | dict[str, Any]
    ],
) -> list[Any] | tuple[Any]
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
def parse_function_params(
    params: Optional[list[Any] | tuple[Any] | dict[str, Any]]
) -> list[Any] | tuple[Any]:
    params = unwrap_generic_value(params)
    if params is None:
        return []
    if isinstance(params, list):
        return [unwrap_generic_value(i) for i in params]
    if isinstance(params, tuple):
        return tuple(unwrap_generic_value(i) for i in params)
    if isinstance(params, dict):
        return list(params.values())
    raise ValueError(f"Invalid type for function parameters: {type(params)}")

call_function_on_contract 

call_function_on_contract(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
) -> Any
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@tenacity.retry(
    wait=tenacity.wait_chain(*[tenacity.wait_fixed(n) for n in range(1, 6)]),
    stop=tenacity.stop_after_attempt(5),
    after=lambda x: logger.debug(
        f"call_function_on_contract failed, {x.attempt_number=}."
    ),
)
def call_function_on_contract(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    function_name: str,
    function_params: Optional[list[Any] | dict[str, Any]] = None,
) -> Any:
    contract = web3.eth.contract(address=contract_address, abi=contract_abi)
    output = contract.functions[function_name](*parse_function_params(function_params)).call()  # type: ignore # TODO: Fix Mypy, as this works just OK.
    return output

prepare_tx 

prepare_tx(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    from_address: ChecksumAddress | None,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    access_list: Optional[AccessList] = None,
    tx_params: Optional[TxParams] = None,
) -> TxParams
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def prepare_tx(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    from_address: ChecksumAddress | None,
    function_name: str,
    function_params: Optional[list[Any] | dict[str, Any]] = None,
    access_list: Optional[AccessList] = None,
    tx_params: Optional[TxParams] = None,
) -> TxParams:
    tx_params_new = _prepare_tx_params(web3, from_address, access_list, tx_params)
    contract = web3.eth.contract(address=contract_address, abi=contract_abi)

    # Build the transaction.
    function_call = contract.functions[function_name](*parse_function_params(function_params))  # type: ignore # TODO: Fix Mypy, as this works just OK.
    built_tx_params: TxParams = function_call.build_transaction(tx_params_new)
    return built_tx_params

send_function_on_contract_tx 

send_function_on_contract_tx(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    from_private_key: PrivateKey,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@tenacity.retry(
    # Don't retry on `reverted` messages, as they would always fail again.
    # TODO: Check this, see https://github.com/gnosis/prediction-market-agent-tooling/issues/625.
    # retry=tenacity.retry_if_exception_message(match=NOT_REVERTED_ICASE_REGEX_PATTERN),
    wait=tenacity.wait_chain(*[tenacity.wait_fixed(n) for n in range(1, 6)]),
    stop=tenacity.stop_after_attempt(5),
    after=lambda x: logger.debug(
        f"send_function_on_contract_tx failed, {x.attempt_number=}."
    ),
)
def send_function_on_contract_tx(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    from_private_key: PrivateKey,
    function_name: str,
    function_params: Optional[list[Any] | dict[str, Any]] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
) -> TxReceipt:
    public_key = private_key_to_public_key(from_private_key)

    tx_params = prepare_tx(
        web3=web3,
        contract_address=contract_address,
        contract_abi=contract_abi,
        from_address=public_key,
        function_name=function_name,
        function_params=function_params,
        tx_params=tx_params,
    )

    receipt_tx = sign_send_and_get_receipt_tx(
        web3, tx_params, from_private_key, timeout
    )
    return receipt_tx

send_function_on_contract_tx_using_safe 

send_function_on_contract_tx_using_safe(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    from_private_key: PrivateKey,
    safe_address: ChecksumAddress,
    function_name: str,
    function_params: Optional[
        list[Any] | dict[str, Any]
    ] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
@tenacity.retry(
    # Don't retry on `reverted` messages, as they would always fail again.
    # TODO: Check this, see https://github.com/gnosis/prediction-market-agent-tooling/issues/625.
    # retry=tenacity.retry_if_exception_message(match=NOT_REVERTED_ICASE_REGEX_PATTERN),
    wait=tenacity.wait_chain(*[tenacity.wait_fixed(n) for n in range(1, 6)]),
    stop=tenacity.stop_after_attempt(5),
    after=lambda x: logger.debug(
        f"send_function_on_contract_tx_using_safe failed, {x.attempt_number=}."
    ),
)
def send_function_on_contract_tx_using_safe(
    web3: Web3,
    contract_address: ChecksumAddress,
    contract_abi: ABI,
    from_private_key: PrivateKey,
    safe_address: ChecksumAddress,
    function_name: str,
    function_params: Optional[list[Any] | dict[str, Any]] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
) -> TxReceipt:
    if not web3.provider.endpoint_uri:  # type: ignore
        raise EnvironmentError("RPC_URL not available in web3 object.")
    ethereum_client = EthereumClient(ethereum_node_url=URI(web3.provider.endpoint_uri))  # type: ignore
    s = SafeV141(safe_address, ethereum_client)
    safe_master_copy_address = s.retrieve_master_copy_address()
    eoa_public_key = private_key_to_public_key(from_private_key)
    # See https://ethereum.stackexchange.com/questions/123750/how-to-implement-eip-2930-access-list for details,
    # required to not go out-of-gas when calling a contract functions using Safe.
    access_list = AccessList(
        [
            AccessListEntry(
                {
                    "address": eoa_public_key,
                    "storageKeys": [HASH_ZERO],
                }
            ),
            AccessListEntry(
                {
                    "address": safe_address,
                    "storageKeys": [HASH_ZERO],
                }
            ),
            AccessListEntry(
                {
                    "address": safe_master_copy_address,
                    "storageKeys": [],
                }
            ),
        ]
    )
    tx_params = prepare_tx(
        web3=web3,
        contract_address=contract_address,
        contract_abi=contract_abi,
        from_address=safe_address,
        function_name=function_name,
        function_params=function_params,
        access_list=access_list,
        tx_params=tx_params,
    )
    safe_tx = s.build_multisig_tx(
        to=Web3.to_checksum_address(tx_params["to"]),
        data=HexBytes(tx_params["data"]),
        value=tx_params["value"],
    )
    safe_tx.sign(from_private_key.get_secret_value())
    safe_tx.call()  # simulate call
    eoa_nonce = web3.eth.get_transaction_count(eoa_public_key)
    tx_hash, tx = safe_tx.execute(
        from_private_key.get_secret_value(),
        tx_nonce=eoa_nonce,
    )
    receipt_tx = web3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout)
    check_tx_receipt(receipt_tx)
    return receipt_tx

sign_send_and_get_receipt_tx 

sign_send_and_get_receipt_tx(
    web3: Web3,
    tx_params_new: TxParams,
    from_private_key: PrivateKey,
    timeout: int = 180,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def sign_send_and_get_receipt_tx(
    web3: Web3,
    tx_params_new: TxParams,
    from_private_key: PrivateKey,
    timeout: int = 180,
) -> TxReceipt:
    # Sign with the private key.
    signed_tx = web3.eth.account.sign_transaction(
        tx_params_new, private_key=from_private_key.get_secret_value()
    )
    # Send the signed transaction.
    send_tx = web3.eth.send_raw_transaction(signed_tx.rawTransaction)
    # And wait for the receipt.
    receipt_tx = web3.eth.wait_for_transaction_receipt(send_tx, timeout=timeout)
    # Verify it didn't fail.
    check_tx_receipt(receipt_tx)
    return receipt_tx

send_xdai_to 

send_xdai_to(
    web3: Web3,
    from_private_key: PrivateKey,
    to_address: ChecksumAddress,
    value: xDaiWei,
    data_text: Optional[str | bytes] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
) -> TxReceipt
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def send_xdai_to(
    web3: Web3,
    from_private_key: PrivateKey,
    to_address: ChecksumAddress,
    value: xDaiWei,
    data_text: Optional[str | bytes] = None,
    tx_params: Optional[TxParams] = None,
    timeout: int = 180,
) -> TxReceipt:
    from_address = private_key_to_public_key(from_private_key)

    tx_params_new: TxParams = {"value": value.value, "to": to_address}
    if data_text is not None:
        tx_params_new["data"] = (
            Web3.to_bytes(text=data_text)
            if not isinstance(data_text, bytes)
            else data_text
        )

    if tx_params:
        tx_params_new.update(tx_params)

    latest_nonce = web3.eth.get_transaction_count(from_address, "pending")
    tx_params_new["nonce"] = latest_nonce

    current_gas_price = web3.eth.gas_price
    tx_params_new["gasPrice"] = int(current_gas_price * 1.2)  # Increase gas by 20%

    gas_estimate = web3.eth.estimate_gas(tx_params_new)
    tx_params_new["gas"] = int(gas_estimate * 1.5)

    receipt_tx = sign_send_and_get_receipt_tx(
        web3, tx_params_new, from_private_key, timeout
    )
    return receipt_tx

ipfscidv0_to_byte32 

ipfscidv0_to_byte32(cid: IPFSCIDVersion0) -> HexBytes

Convert ipfscidv0 to 32 bytes. Modified from https://github.com/emg110/ipfs2bytes32/blob/main/python/ipfs2bytes32.py

Source code in prediction_market_agent_tooling/tools/web3_utils.py 
333
334
335
336
337
338
339
340
def ipfscidv0_to_byte32(cid: IPFSCIDVersion0) -> HexBytes:
    """
    Convert ipfscidv0 to 32 bytes.
    Modified from https://github.com/emg110/ipfs2bytes32/blob/main/python/ipfs2bytes32.py
    """
    decoded = base58.b58decode(cid)
    sliced_decoded = decoded[2:]
    return HexBytes(binascii.b2a_hex(sliced_decoded).decode("utf-8"))

byte32_to_ipfscidv0 

byte32_to_ipfscidv0(hex: HexBytes) -> IPFSCIDVersion0

Convert 32 bytes hex to ipfscidv0. Modified from https://github.com/emg110/ipfs2bytes32/blob/main/python/ipfs2bytes32.py

Source code in prediction_market_agent_tooling/tools/web3_utils.py 
343
344
345
346
347
348
349
def byte32_to_ipfscidv0(hex: HexBytes) -> IPFSCIDVersion0:
    """
    Convert 32 bytes hex to ipfscidv0.
    Modified from https://github.com/emg110/ipfs2bytes32/blob/main/python/ipfs2bytes32.py
    """
    completed_binary_str = b"\x12 " + hex
    return IPFSCIDVersion0(base58.b58encode(completed_binary_str).decode("utf-8"))

get_receipt_block_timestamp 

get_receipt_block_timestamp(
    receipt_tx: TxReceipt, web3: Web3
) -> int
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
352
353
354
355
356
357
358
359
360
361
362
363
@tenacity.retry(
    wait=tenacity.wait_chain(*[tenacity.wait_fixed(n) for n in range(1, 6)]),
    stop=tenacity.stop_after_attempt(5),
    after=lambda x: logger.debug(
        f"get_receipt_block_timestamp failed, {x.attempt_number=}."
    ),
)
def get_receipt_block_timestamp(receipt_tx: TxReceipt, web3: Web3) -> int:
    block_number = receipt_tx["blockNumber"]
    block = web3.eth.get_block(block_number)
    block_timestamp: int = block["timestamp"]
    return block_timestamp

is_valid_wei 

is_valid_wei(value: Web3Wei) -> bool
Source code in prediction_market_agent_tooling/tools/web3_utils.py 
366
367
def is_valid_wei(value: Web3Wei) -> bool:
    return MIN_WEI <= value <= MAX_WEI