Feature branch for WIP microchain agent

prediction_market_agent/agents/microchain_agent/functions.py

@@ -0,0 +1,224 @@
+import pprint
+import typing as t
+from decimal import Decimal
+
+from microchain import Function
+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}
+outcomeTokens["Will Bitcoin hit 100k in 2024?"] = {"yes": 0, "no": 0}
+
+
+class Sum(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to compute the sum of two numbers"
+
+    @property
+    def example_args(self) -> list[float]:
+        return [2, 2]
+
+    def __call__(self, a: float, b: float) -> float:
+        return a + b
+
+
+class Product(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to compute the product of two numbers"
+
+    @property
+    def example_args(self) -> list[float]:
+        return [2, 2]
+
+    def __call__(self, a: float, b: float) -> float:
+        return a * b
+
+
+class GetMarkets(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to get a list of predction markets and the current yes prices"
+
+    @property
+    def example_args(self) -> list[str]:
+        return []
+
+    def __call__(self) -> list[str]:
+        return [
+            str(MicroMarket.from_agent_market(m)) for m in get_omen_binary_markets()
+        ]
+
+
+class GetPropabilityForQuestion(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to research the probability of an event occuring"
+
+    @property
+    def example_args(self) -> list[str]:
+        return ["Will Joe Biden get reelected in 2024?"]
+
+    def __call__(self, a: str) -> float:
+        if a == "Will Joe Biden get reelected in 2024?":
+            return 0.41
+        if a == "Will Bitcoin hit 100k in 2024?":
+            return 0.22
+
+        return 0.0
+
+
+class GetBalance(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to get your own balance in $"
+
+    @property
+    def example_args(self) -> list[str]:
+        return []
+
+    def __call__(self) -> float:
+        print(f"Your balance is: {balance} and ")
+        pprint.pprint(outcomeTokens)
+        return balance
+
+
+class BuyTokens(Function):
+    def __init__(self, outcome: str):
+        self.outcome = outcome
+        super().__init__()
+
+    @property
+    def description(self) -> str:
+        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.3]
+
+    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 BuyYes(BuyTokens):
+    def __init__(self) -> None:
+        super().__init__(OMEN_TRUE_OUTCOME)
+
+
+class BuyNo(BuyTokens):
+    def __init__(self) -> None:
+        super().__init__(OMEN_FALSE_OUTCOME)
+
+
+class SellYes(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to sell yes outcome tokens of a prediction market. The second parameter specifies how much tokens you sell."
+
+    @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 outcomeTokens
+        if amount > outcomeTokens[market]["yes"]:
+            return f"Your balance of {outcomeTokens[market]['yes']} yes outcome tokens is not large enough to sell {amount}."
+
+        outcomeTokens[market]["yes"] -= amount
+        return "Sold " + str(amount) + " yes outcome token of: " + market
+
+
+class SellNo(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to sell no outcome tokens of a prdiction market. The second parameter specifies how much tokens you sell."
+
+    @property
+    def example_args(self) -> list[t.Union[str, float]]:
+        return ["Will Joe Biden get reelected in 2024?", 4]
+
+    def __call__(self, market: str, amount: int) -> str:
+        global outcomeTokens
+        if amount > outcomeTokens[market]["no"]:
+            return f"Your balance of {outcomeTokens[market]['no']} no outcome tokens is not large enough to sell {amount}."
+
+        outcomeTokens[market]["no"] -= amount
+        return "Sold " + str(amount) + " no outcome token of: " + market
+
+
+class BalanceToOutcomes(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function to convert your balance into equal units of 'yes' and 'no' outcome tokens. The function takes the amount of balance as the argument."
+
+    @property
+    def example_args(self) -> list[t.Union[str, float]]:
+        return ["Will Joe Biden get reelected in 2024?", 50]
+
+    def __call__(self, market: str, amount: int) -> str:
+        global balance
+        global outcomeTokens
+        outcomeTokens[market]["yes"] += amount
+        outcomeTokens[market]["no"] += amount
+        balance -= amount
+        return f"Converted {amount} units of balance into {amount} 'yes' outcome tokens and {amount} 'no' outcome tokens. Remaining balance is {balance}."
+
+
+class SummarizeLearning(Function):
+    @property
+    def description(self) -> str:
+        return "Use this function summarize your learnings and save them so that you can access them later."
+
+    @property
+    def example_args(self) -> list[str]:
+        return [
+            "Today I learned that I need to check my balance fore making decisions about how much to invest."
+        ]
+
+    def __call__(self, summary: str) -> str:
+        # print(summary)
+        # pprint.pprint(outcomeTokens)
+        return summary
+
+
+ALL_FUNCTIONS = [
+    Sum,
+    Product,
+    GetMarkets,
+    GetPropabilityForQuestion,
+    GetBalance,
+    BuyYes,
+    BuyNo,
+    SellYes,
+    SellNo,
+    # BalanceToOutcomes,
+    SummarizeLearning,
+]

prediction_market_agent/agents/microchain_agent/microchain_agent.py

@@ -0,0 +1,34 @@
+import os
+
+from dotenv import load_dotenv
+from functions import ALL_FUNCTIONS
+from microchain import LLM, Agent, Engine, OpenAIChatGenerator
+from microchain.functions import Reasoning, Stop
+
+load_dotenv()
+
+engine = Engine()
+engine.register(Reasoning())
+engine.register(Stop())
+for function in ALL_FUNCTIONS:
+    engine.register(function())
+
+generator = OpenAIChatGenerator(
+    model="gpt-4-turbo-preview",
+    api_key=os.getenv("OPENAI_API_KEY"),
+    api_base="https://api.openai.com/v1",
+    temperature=0.7,
+)
+agent = Agent(llm=LLM(generator=generator), engine=engine)
+agent.prompt = f"""Act as a agent. You can use the following functions:
+
+{engine.help}
+
+
+Only output valid Python function calls.
+
+"""
+
+agent.bootstrap = ['Reasoning("I need to reason step-by-step")']
+agent.run(iterations=3)
+generator.print_usage()

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

pyproject.toml

@@ -41,6 +41,8 @@ autoflake = "^2.2.1"
 isort = "^5.13.2"
 markdownify = "^0.11.6"
 tavily-python = "^0.3.1"
+# TODO remove when PR is merged
+microchain-python = {git = "https://github.com/evangriffiths/microchain.git", rev = "evan/token-tracker"}
 
 [build-system]
 requires = ["poetry-core"]

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"