Implement BuyYes and BuyNo functions with Omen API

prediction_market_agent/agents/microchain_agent/functions.py

@@ -1,14 +1,23 @@
 import pprint
 import typing as t
+from decimal import Decimal
 
 from microchain import Function
-from prediction_market_agent_tooling.markets.agent_market import (
-    AgentMarket,
-    FilterBy,
-    SortBy,
+from prediction_market_agent_tooling.markets.data_models import BetAmount, Currency
+from prediction_market_agent_tooling.markets.omen.data_models import (
+    OMEN_FALSE_OUTCOME,
+    OMEN_TRUE_OUTCOME,
+    get_boolean_outcome,
 )
 from prediction_market_agent_tooling.markets.omen.omen import OmenAgentMarket
 
+from prediction_market_agent.agents.microchain_agent.utils import (
+    MicroMarket,
+    get_omen_binary_market_from_question,
+    get_omen_binary_markets,
+    get_omen_market_token_balance,
+)
+
 balance = 50
 outcomeTokens = {}
 outcomeTokens["Will Joe Biden get reelected in 2024?"] = {"yes": 0, "no": 0}
@@ -51,18 +60,9 @@ def example_args(self) -> list[str]:
         return []
 
     def __call__(self) -> list[str]:
-        # Get the 5 markets that are closing soonest
-        markets: list[AgentMarket] = OmenAgentMarket.get_binary_markets(
-            filter_by=FilterBy.OPEN,
-            sort_by=SortBy.CLOSING_SOONEST,
-            limit=5,
-        )
-
-        market_questions_and_prices = []
-        for market in markets:
-            market_questions_and_prices.append(market.question)
-            market_questions_and_prices.append(str(market.p_yes))
-        return market_questions_and_prices
+        return [
+            str(MicroMarket.from_agent_market(m)) for m in get_omen_binary_markets()
+        ]
 
 
 class GetPropabilityForQuestion(Function):
@@ -98,44 +98,44 @@ def __call__(self) -> float:
         return balance
 
 
-class BuyYes(Function):
+class BuyTokens(Function):
+    def __init__(self, outcome: str):
+        self.outcome = outcome
+        super().__init__()
+
     @property
     def description(self) -> str:
-        return "Use this function to buy yes outcome tokens of a prediction market. The second parameter specifies how much $ you spend."
+        return f"Use this function to buy {self.outcome} outcome tokens of a prediction market. The second parameter specifies how much $ you spend."
 
     @property
     def example_args(self) -> list[t.Union[str, float]]:
-        return ["Will Joe Biden get reelected in 2024?", 2]
-
-    def __call__(self, market: str, amount: int) -> str:
-        global balance
-        if amount > balance:
-            return (
-                f"Your balance of {balance} $ is not large enough to spend {amount} $."
-            )
+        return ["Will Joe Biden get reelected in 2024?", 2.3]
 
-        balance -= amount
-        return "Bought " + str(amount * 2) + " yes outcome token of: " + market
+    def __call__(self, market: str, amount: float) -> str:
+        outcome_bool = get_boolean_outcome(self.outcome)
 
+        market_obj: OmenAgentMarket = get_omen_binary_market_from_question(market)
+        before_balance = get_omen_market_token_balance(
+            market=market_obj, outcome=outcome_bool
+        )
+        market_obj.place_bet(
+            outcome_bool, BetAmount(amount=Decimal(amount), currency=Currency.xDai)
+        )
+        tokens = (
+            get_omen_market_token_balance(market=market_obj, outcome=outcome_bool)
+            - before_balance
+        )
+        return f"Bought {tokens} {self.outcome} outcome tokens of: {market}"
 
-class BuyNo(Function):
-    @property
-    def description(self) -> str:
-        return "Use this function to buy no outcome tokens of a prdiction market. The second parameter specifies how much $ you spend."
 
-    @property
-    def example_args(self) -> list[t.Union[str, float]]:
-        return ["Will Joe Biden get reelected in 2024?", 4]
+class BuyYes(BuyTokens):
+    def __init__(self) -> None:
+        super().__init__(OMEN_TRUE_OUTCOME)
 
-    def __call__(self, market: str, amount: int) -> str:
-        global balance
-        if amount > balance:
-            return (
-                f"Your balance of {balance} $ is not large enough to spend {amount} $."
-            )
 
-        balance -= amount
-        return "Bought " + str(amount * 2) + " no outcome token of: " + market
+class BuyNo(BuyTokens):
+    def __init__(self) -> None:
+        super().__init__(OMEN_FALSE_OUTCOME)
 
 
 class SellYes(Function):

prediction_market_agent/agents/microchain_agent/utils.py

@@ -0,0 +1,47 @@
+from typing import List, cast
+
+from prediction_market_agent_tooling.markets.agent_market import (
+    AgentMarket,
+    FilterBy,
+    SortBy,
+)
+from prediction_market_agent_tooling.markets.omen.omen import OmenAgentMarket
+from pydantic import BaseModel
+
+
+class MicroMarket(BaseModel):
+    question: str
+    p_yes: float
+
+    @staticmethod
+    def from_agent_market(market: OmenAgentMarket) -> "MicroMarket":
+        return MicroMarket(
+            question=market.question,
+            p_yes=float(market.p_yes),
+        )
+
+    def __str__(self) -> str:
+        return f"'{self.question}' with probability of yes: {self.p_yes:.2%}"
+
+
+def get_omen_binary_markets() -> list[OmenAgentMarket]:
+    # Get the 5 markets that are closing soonest
+    markets: list[AgentMarket] = OmenAgentMarket.get_binary_markets(
+        filter_by=FilterBy.OPEN,
+        sort_by=SortBy.CLOSING_SOONEST,
+        limit=5,
+    )
+    return cast(List[OmenAgentMarket], markets)
+
+
+def get_omen_binary_market_from_question(market: str) -> OmenAgentMarket:
+    markets = get_omen_binary_markets()
+    for m in markets:
+        if m.question == market:
+            return m
+    raise ValueError(f"Market '{market}' not found")
+
+
+def get_omen_market_token_balance(market: OmenAgentMarket, outcome: bool) -> float:
+    # TODO implement this
+    return 7.3

tests/agents/test_microchain.py

@@ -0,0 +1,30 @@
+import pytest
+
+from prediction_market_agent.agents.microchain_agent.functions import (
+    BuyNo,
+    BuyYes,
+    GetMarkets,
+)
+from prediction_market_agent.agents.microchain_agent.utils import (
+    get_omen_binary_markets,
+)
+from tests.utils import RUN_PAID_TESTS
+
+
+def test_get_markets() -> None:
+    get_markets = GetMarkets()
+    assert len(get_markets()) > 0
+
+
+@pytest.mark.skipif(not RUN_PAID_TESTS, reason="This test costs money to run.")
+def test_buy_yes() -> None:
+    market = get_omen_binary_markets()[0]
+    buy_yes = BuyYes()
+    print(buy_yes(market.question, 0.0001))
+
+
+@pytest.mark.skipif(not RUN_PAID_TESTS, reason="This test costs money to run.")
+def test_buy_no() -> None:
+    market = get_omen_binary_markets()[0]
+    buy_yes = BuyNo()
+    print(buy_yes(market.question, 0.0001))

tests/utils.py

@@ -0,0 +1,3 @@
+import os
+
+RUN_PAID_TESTS = os.environ.get("RUN_PAID_TESTS", "0") == "1"