mm-fok-sample-integration

modules/sample_market_maker_liquidity_module.py

@@ -0,0 +1,96 @@
+import aiohttp
+import asyncio
+from templates.liquidity_module import Token
+from typing import Dict, Tuple, Literal
+from decimal import Decimal
+
+class MarketMakerLiquidityModule:
+
+    async def get_sell_quote(
+        fixed_parameters: Dict,
+        input_token: Token,
+        output_token: Token,
+        input_amount: int,
+        block: Literal['latest', int] = 'latest'
+    ) -> Tuple[int | None, int | None]:
+
+        market_maker_api = fixed_parameters.get("market_maker_api")
+        api_key = fixed_parameters.get("api_key", None)
+        chain = fixed_parameters.get("chain", "ethereum")
+        user_address = fixed_parameters.get("user_address")
+
+        # Construct the URL for the API call
+        url = f"{market_maker_api}/sellQuote"
+
+        body = {
+            "chain": chain,
+            "sell_token": input_token.address,
+            "buy_token": output_token.address,
+            "sell_amounts": input_amount,
+            "user_address": user_address
+        }
+
+
+        headers = {
+            "api-key": api_key,
+        }
+
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(url, headers=headers, json=body) as resp:
+                    data = await resp.json()
+
+                    if "SUCCESS" not in data.get("status", ""):
+                        return None, None
+
+                    output_amount = data.get("buy_amount")
+
+                    return 0, output_amount
+
+            except Exception as e:
+                print(f"Error fetching sell quote: {e}")
+                return None, None
+
+    async def get_buy_quote(
+        fixed_parameters: Dict,
+        input_token: Token,
+        output_token: Token,
+        output_amount: int,
+        block: Literal['latest', int] = 'latest'
+    ) -> Tuple[int | None, int | None]:
+
+        market_maker_api = fixed_parameters.get("market_maker_api")
+        api_key = fixed_parameters.get("api_key", None)
+        chain = fixed_parameters.get("chain", "ethereum")
+        user_address = fixed_parameters.get("user_address")
+
+        # Construct the URL for the API call
+        url = f"{market_maker_api}/buyQuote"
+
+        body = {
+            "chain": chain,
+            "buy_token": output_token.address,
+            "sell_token": input_token.address,
+            "buy_amount": output_amount,
+            "user_address": user_address
+        }
+
+        headers = {
+            "api-key": api_key,
+        }
+
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(url, headers=headers, json=body) as resp:
+                    data = await resp.json()
+
+                    if "SUCCESS" not in data.get("status", ""):
+                        return None, None
+
+                    input_amount = data.get("sell_amount")
+
+                    return 0, input_amount
+
+            except Exception as e:
+                print(f"Error fetching buy quote: {e}")
+                return None, None

tests/test_sample_market_maker_liquidity_module.py

@@ -0,0 +1,81 @@
+import unittest
+from unittest.mock import patch, AsyncMock
+import sys
+from os import listdir, path, walk
+
+sys.path.append(path.dirname(path.dirname(path.abspath(__file__))))
+sys.path.append(path.dirname(path.dirname(path.dirname(path.abspath(__file__)))))
+sys.path.append(path.dirname(path.dirname(path.dirname(path.dirname(path.abspath(__file__))))))
+sys.path.append(path.dirname(path.dirname(path.dirname(path.dirname(path.dirname(path.abspath(__file__)))))))
+sys.path.append(path.dirname(path.dirname(path.dirname(path.dirname(path.dirname(path.dirname(path.abspath(__file__))))))))
+from decimal import Decimal
+from templates.liquidity_module import Token
+from modules.sample_market_maker_liquidity_module import MarketMakerLiquidityModule  # Replace with the actual module name
+
+class TestMarketMakerLiquidityModule(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.fixed_parameters = {
+            "market_maker_api": "https://mock.api",
+            "api_key": "test-api-key",
+            "chain": "ethereum",
+            "user_address": "0xuser"
+        }
+        self.input_token = Token(address="0xinput", symbol="InputToken", decimals=18, reference_price=Decimal("0.0001"))
+        self.output_token = Token(address="0xoutput", symbol="OutputToken", decimals=18, reference_price=Decimal("0.0002"))
+
+    @patch("aiohttp.ClientSession.post")
+    async def test_get_sell_quote_success(self, mock_post):
+        mock_response = AsyncMock()
+        mock_response.json = AsyncMock(return_value={
+            "status": "SUCCESS",
+            "buy_amount": 1000
+        })
+        mock_post.return_value.__aenter__.return_value = mock_response
+
+        fee, amount = await MarketMakerLiquidityModule.get_sell_quote(
+            self.fixed_parameters, self.input_token, self.output_token, 500
+        )
+
+        self.assertEqual(fee, 0)
+        self.assertEqual(amount, 1000)
+
+    @patch("aiohttp.ClientSession.post")
+    async def test_get_sell_quote_failure(self, mock_post):
+        mock_response = AsyncMock()
+        mock_response.json = AsyncMock(return_value={"status": "FAILED"})
+        mock_post.return_value.__aenter__.return_value = mock_response
+
+        fee, amount = await MarketMakerLiquidityModule.get_sell_quote(
+            self.fixed_parameters, self.input_token, self.output_token, 500
+        )
+        self.assertIsNone(fee)
+        self.assertIsNone(amount)
+
+    @patch("aiohttp.ClientSession.post")
+    async def test_get_buy_quote_success(self, mock_post):
+        mock_response = AsyncMock()
+        mock_response.json = AsyncMock(return_value={
+            "status": "SUCCESS",
+            "sell_amount": 800
+        })
+        mock_post.return_value.__aenter__.return_value = mock_response
+
+        fee, amount = await MarketMakerLiquidityModule.get_buy_quote(
+            self.fixed_parameters, self.input_token, self.output_token, 1000
+        )
+        self.assertEqual(fee, 0)
+        self.assertEqual(amount, 800)
+
+    @patch("aiohttp.ClientSession.post")
+    async def test_get_buy_quote_failure(self, mock_post):
+        mock_response = AsyncMock()
+        mock_response.json = AsyncMock(return_value={"status": "ERROR"})
+        mock_post.return_value.__aenter__.return_value = mock_response
+
+        fee, amount = await MarketMakerLiquidityModule.get_buy_quote(
+            self.fixed_parameters, self.input_token, self.output_token, 1000
+        )
+        self.assertIsNone(fee)
+        self.assertIsNone(amount)
+
+unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(TestMarketMakerLiquidityModule))