diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f0f08f0 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +USER_ADDRESS= +PRIVATE_KEY= diff --git a/cow_py/order_book/__init__.py b/cow_py/order_book/__init__.py index acdaade..e69de29 100644 --- a/cow_py/order_book/__init__.py +++ b/cow_py/order_book/__init__.py @@ -1,243 +0,0 @@ -from dataclasses import asdict -from typing import Any, Dict, List -from cow_py.common.config import DEFAULT_COW_API_CONTEXT, CowEnv, SupportedChainId -from cow_py.order_book.requests import DEFAULT_BACKOFF_OPTIONS, request -from .generated.model import ( - AppDataObject, - Trade, - Order, - TotalSurplus, - NativePriceResponse, - SolverCompetitionResponse, - OrderQuoteRequest, - OrderQuoteResponse, - OrderCreation, - UID, - Address, - TransactionHash, - AppDataHash, - OrderCancellation, -) - -ORDER_BOOK_PROD_CONFIG = { - SupportedChainId.MAINNET: "https://api.cow.fi/mainnet", - SupportedChainId.GNOSIS_CHAIN: "https://api.cow.fi/xdai", - SupportedChainId.SEPOLIA: "https://api.cow.fi/sepolia", -} - -ORDER_BOOK_STAGING_CONFIG = { - SupportedChainId.MAINNET: "https://barn.api.cow.fi/mainnet", - SupportedChainId.GNOSIS_CHAIN: "https://barn.api.cow.fi/xdai", - SupportedChainId.SEPOLIA: "https://barn.api.cow.fi/sepolia", -} - - -class OrderBookApi: - def __init__(self, context: Dict[str, Any] = None): - if context is None: - context = {} - self.context = { - **asdict(DEFAULT_COW_API_CONTEXT), - "backoffOpts": DEFAULT_BACKOFF_OPTIONS, - **context, - } - - def get_api_url(self, context: Dict[str, Any]) -> str: - if context.get("env", CowEnv.PROD) == CowEnv.PROD: - return ORDER_BOOK_PROD_CONFIG[ - context.get("chain_id", SupportedChainId.MAINNET) - ] - return ORDER_BOOK_STAGING_CONFIG[ - context.get("chain_id", SupportedChainId.MAINNET) - ] - - async def _fetch( - self, - path: str, - context_override: Dict[str, Any] = None, - **request_kwargs, - ) -> Any: - context = self._get_context_with_override(context_override) - url = self.get_api_url(context) - backoff_opts = context.get("backoffOpts", DEFAULT_BACKOFF_OPTIONS) - return await request( - url, path=path, backoff_opts=backoff_opts, **request_kwargs - ) - - def _get_context_with_override( - self, context_override: Dict[str, Any] = None - ) -> Dict[str, Any]: - if context_override is None: - context_override = {} - return {**self.context, **context_override} - - async def get_version(self, context_override: Dict[str, Any] = None) -> str: - return await self._fetch( - path="/api/v1/version", context_override=context_override - ) - - async def get_trades_by_owner( - self, owner: Address, context_override: Dict[str, Any] = None - ) -> List[Trade]: - response = await self._fetch( - path="/api/v1/trades", - params={"owner": owner}, - context_override=context_override, - ) - return [Trade(**trade) for trade in response] - - async def get_trades_by_order_uid( - self, order_uid: UID, context_override: Dict[str, Any] = None - ) -> List[Trade]: - response = await self._fetch( - path="/api/v1/trades", - params={"order_uid": order_uid}, - context_override=context_override, - ) - return [Trade(**trade) for trade in response] - - async def get_orders_by_owner( - self, - owner: Address, - limit: int = 1000, - offset: int = 0, - context_override: Dict[str, Any] = None, - ) -> List[Order]: - return [ - Order(**order) - for order in await self._fetch( - path=f"/api/v1/account/{owner}/orders", - params={"limit": limit, "offset": offset}, - context_override=context_override, - ) - ] - - async def get_order_by_uid( - self, order_uid: UID, context_override: Dict[str, Any] = None - ) -> Order: - response = await self._fetch( - path=f"/api/v1/orders/{order_uid}", - context_override=context_override, - ) - return Order(**response) - - def get_order_link( - self, order_uid: UID, context_override: Dict[str, Any] = None - ) -> str: - return ( - self.get_api_url(self._get_context_with_override(context_override)) - + f"/orders/{order_uid}" - ) - - async def get_tx_orders( - self, tx_hash: TransactionHash, context_override: Dict[str, Any] = None - ) -> List[Order]: - response = await self._fetch( - path=f"/api/v1/transactions/{tx_hash}/orders", - context_override=context_override, - ) - return [Order(**order) for order in response] - - async def get_native_price( - self, tokenAddress: Address, context_override: Dict[str, Any] = None - ) -> NativePriceResponse: - response = await self._fetch( - path=f"/api/v1/token/{tokenAddress}/native_price", - context_override=context_override, - ) - return NativePriceResponse(**response) - - async def get_total_surplus( - self, user: Address, context_override: Dict[str, Any] = None - ) -> TotalSurplus: - response = await self._fetch( - path=f"/api/v1/users/{user}/total_surplus", - context_override=context_override, - ) - return TotalSurplus(**response) - - async def get_app_data( - self, app_data_hash: AppDataHash, context_override: Dict[str, Any] = None - ) -> Dict[str, Any]: - return await self._fetch( - path=f"/api/v1/app_data/{app_data_hash}", - context_override=context_override, - ) - - async def get_solver_competition( - self, action_id: int = "latest", context_override: Dict[str, Any] = None - ) -> SolverCompetitionResponse: - response = await self._fetch( - path=f"/api/v1/solver_competition/{action_id}", - context_override=context_override, - ) - return SolverCompetitionResponse(**response) - - async def get_solver_competition_by_tx_hash( - self, tx_hash: TransactionHash, context_override: Dict[str, Any] = None - ) -> SolverCompetitionResponse: - response = await self._fetch( - path=f"/api/v1/solver_competition/by_tx_hash/{tx_hash}", - context_override=context_override, - ) - return SolverCompetitionResponse(**response) - - async def get_quote( - self, request: OrderQuoteRequest, context_override: Dict[str, Any] = None - ) -> OrderQuoteResponse: - response = await self._fetch( - path="/api/v1/quote", - json=request.dict(), - context_override=context_override, - ) - return OrderQuoteResponse(**response) - - async def post_quote( - self, request: OrderQuoteRequest, context_override: Dict[str, Any] = None - ) -> OrderQuoteResponse: - response = await self._fetch( - path="/api/v1/quote", - json=request.dict(), - context_override=context_override, - method="POST", - ) - return OrderQuoteResponse(**response) - - async def post_order( - self, order: OrderCreation, context_override: Dict[str, Any] = None - ): - response = await self._fetch( - path="/api/v1/orders", - json=order.dict(), - context_override=context_override, - method="POST", - ) - return UID(response) - - async def delete_order( - self, - orders_cancelation: OrderCancellation, - context_override: Dict[str, Any] = None, - ): - response = await self._fetch( - path="/api/v1/orders", - json=orders_cancelation.dict(), - context_override=context_override, - method="DELETE", - ) - return UID(response) - - async def put_app_data( - self, - app_data: AppDataObject, - app_data_hash: str = None, - context_override: Dict[str, Any] = None, - ) -> AppDataHash: - app_data_hash_url = app_data_hash if app_data_hash else "" - response = await self._fetch( - path=f"/api/v1/app_data/{app_data_hash_url}", - json=app_data.dict(), - context_override=context_override, - method="PUT", - ) - return AppDataHash(response) diff --git a/cow_py/order_book/api.py b/cow_py/order_book/api.py new file mode 100644 index 0000000..85e5c59 --- /dev/null +++ b/cow_py/order_book/api.py @@ -0,0 +1,246 @@ +from dataclasses import asdict +import json +from typing import Any, Dict, List +from cow_py.common.config import DEFAULT_COW_API_CONTEXT, CowEnv, SupportedChainId +from cow_py.order_book.requests import DEFAULT_BACKOFF_OPTIONS, request +from .generated.model import ( + AppDataObject, + OrderQuoteSide, + OrderQuoteValidity, + OrderQuoteValidity1, + Trade, + Order, + TotalSurplus, + NativePriceResponse, + SolverCompetitionResponse, + OrderQuoteRequest, + OrderQuoteResponse, + OrderCreation, + UID, + Address, + TransactionHash, + AppDataHash, + OrderCancellation, +) + +ORDER_BOOK_PROD_CONFIG = { + SupportedChainId.MAINNET: "https://api.cow.fi/mainnet", + SupportedChainId.GNOSIS_CHAIN: "https://api.cow.fi/xdai", + SupportedChainId.SEPOLIA: "https://api.cow.fi/sepolia", +} + +ORDER_BOOK_STAGING_CONFIG = { + SupportedChainId.MAINNET: "https://barn.api.cow.fi/mainnet", + SupportedChainId.GNOSIS_CHAIN: "https://barn.api.cow.fi/xdai", + SupportedChainId.SEPOLIA: "https://barn.api.cow.fi/sepolia", +} + + +class OrderBookApi: + def __init__(self, context: Dict[str, Any] = None): + if context is None: + context = {} + self.context = { + **asdict(DEFAULT_COW_API_CONTEXT), + "backoffOpts": DEFAULT_BACKOFF_OPTIONS, + **context, + } + + def get_api_url(self, context: Dict[str, Any]) -> str: + if context.get("env", CowEnv.PROD) == CowEnv.PROD: + return ORDER_BOOK_PROD_CONFIG[ + context.get("chain_id", SupportedChainId.MAINNET) + ] + return ORDER_BOOK_STAGING_CONFIG[ + context.get("chain_id", SupportedChainId.MAINNET) + ] + + async def _fetch( + self, + path: str, + context_override: Dict[str, Any] = None, + **request_kwargs, + ) -> Any: + context = self._get_context_with_override(context_override) + url = self.get_api_url(context) + backoff_opts = context.get("backoffOpts", DEFAULT_BACKOFF_OPTIONS) + return await request( + url, path=path, backoff_opts=backoff_opts, **request_kwargs + ) + + def _get_context_with_override( + self, context_override: Dict[str, Any] = None + ) -> Dict[str, Any]: + if context_override is None: + context_override = {} + return {**self.context, **context_override} + + async def get_version(self, context_override: Dict[str, Any] = None) -> str: + return await self._fetch( + path="/api/v1/version", context_override=context_override + ) + + async def get_trades_by_owner( + self, owner: Address, context_override: Dict[str, Any] = None + ) -> List[Trade]: + response = await self._fetch( + path="/api/v1/trades", + params={"owner": owner}, + context_override=context_override, + ) + return [Trade(**trade) for trade in response] + + async def get_trades_by_order_uid( + self, order_uid: UID, context_override: Dict[str, Any] = None + ) -> List[Trade]: + response = await self._fetch( + path="/api/v1/trades", + params={"order_uid": order_uid}, + context_override=context_override, + ) + return [Trade(**trade) for trade in response] + + async def get_orders_by_owner( + self, + owner: Address, + limit: int = 1000, + offset: int = 0, + context_override: Dict[str, Any] = None, + ) -> List[Order]: + return [ + Order(**order) + for order in await self._fetch( + path=f"/api/v1/account/{owner}/orders", + params={"limit": limit, "offset": offset}, + context_override=context_override, + ) + ] + + async def get_order_by_uid( + self, order_uid: UID, context_override: Dict[str, Any] = None + ) -> Order: + response = await self._fetch( + path=f"/api/v1/orders/{order_uid}", + context_override=context_override, + ) + return Order(**response) + + def get_order_link( + self, order_uid: UID, context_override: Dict[str, Any] = None + ) -> str: + return ( + self.get_api_url(self._get_context_with_override(context_override)) + + f"api/v1/orders/{order_uid.root}" + ) + + async def get_tx_orders( + self, tx_hash: TransactionHash, context_override: Dict[str, Any] = None + ) -> List[Order]: + response = await self._fetch( + path=f"/api/v1/transactions/{tx_hash}/orders", + context_override=context_override, + ) + return [Order(**order) for order in response] + + async def get_native_price( + self, tokenAddress: Address, context_override: Dict[str, Any] = None + ) -> NativePriceResponse: + response = await self._fetch( + path=f"/api/v1/token/{tokenAddress}/native_price", + context_override=context_override, + ) + return NativePriceResponse(**response) + + async def get_total_surplus( + self, user: Address, context_override: Dict[str, Any] = None + ) -> TotalSurplus: + response = await self._fetch( + path=f"/api/v1/users/{user}/total_surplus", + context_override=context_override, + ) + return TotalSurplus(**response) + + async def get_app_data( + self, app_data_hash: AppDataHash, context_override: Dict[str, Any] = None + ) -> Dict[str, Any]: + return await self._fetch( + path=f"/api/v1/app_data/{app_data_hash}", + context_override=context_override, + ) + + async def get_solver_competition( + self, action_id: int = "latest", context_override: Dict[str, Any] = None + ) -> SolverCompetitionResponse: + response = await self._fetch( + path=f"/api/v1/solver_competition/{action_id}", + context_override=context_override, + ) + return SolverCompetitionResponse(**response) + + async def get_solver_competition_by_tx_hash( + self, tx_hash: TransactionHash, context_override: Dict[str, Any] = None + ) -> SolverCompetitionResponse: + response = await self._fetch( + path=f"/api/v1/solver_competition/by_tx_hash/{tx_hash}", + context_override=context_override, + ) + return SolverCompetitionResponse(**response) + + async def post_quote( + self, + request: OrderQuoteRequest, + side: OrderQuoteSide, + validity: OrderQuoteValidity = OrderQuoteValidity1(), + context_override: Dict[str, Any] = None, + ) -> OrderQuoteResponse: + response = await self._fetch( + path="/api/v1/quote", + json={ + **request.dict(by_alias=True), + # side object need to be converted to json first to avoid on kind type + **json.loads(side.json()), + **validity.dict(), + }, + context_override=context_override, + method="POST", + ) + return OrderQuoteResponse(**response) + + async def post_order( + self, order: OrderCreation, context_override: Dict[str, Any] = None + ): + response = await self._fetch( + path="/api/v1/orders", + json=json.loads(order.json(by_alias=True)), + context_override=context_override, + method="POST", + ) + return UID(response) + + async def delete_order( + self, + orders_cancelation: OrderCancellation, + context_override: Dict[str, Any] = None, + ): + response = await self._fetch( + path="/api/v1/orders", + json=orders_cancelation.json(), + context_override=context_override, + method="DELETE", + ) + return UID(response) + + async def put_app_data( + self, + app_data: AppDataObject, + app_data_hash: str = None, + context_override: Dict[str, Any] = None, + ) -> AppDataHash: + app_data_hash_url = app_data_hash if app_data_hash else "" + response = await self._fetch( + path=f"/api/v1/app_data/{app_data_hash_url}", + json=app_data.json(), + context_override=context_override, + method="PUT", + ) + return AppDataHash(response) diff --git a/examples/__init__.py b/examples/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/order_posting_e2e.py b/examples/order_posting_e2e.py new file mode 100644 index 0000000..ef7182f --- /dev/null +++ b/examples/order_posting_e2e.py @@ -0,0 +1,118 @@ +# To run this test you will need to fill the .env file with the necessary variables (see .env.example). +# You will also need to have enough funds in you wallet of the sell token to create the order. +# The funds have to already be approved to the CoW Swap Vault Relayer + +import asyncio +from dataclasses import asdict +import json +import os +from cow_py.common.chains import Chain +from cow_py.common.config import SupportedChainId +from cow_py.common.constants import CowContractAddress +from cow_py.contracts.domain import domain +from cow_py.contracts.sign import ( + EcdsaSignature, + SigningScheme, + sign_order as _sign_order, +) +from web3 import Account + + +from cow_py.contracts.order import Order +from cow_py.order_book.api import OrderBookApi +from cow_py.order_book.generated.model import ( + UID, + OrderCreation, + OrderQuoteRequest, + OrderQuoteResponse, + OrderQuoteSide, +) + + +BUY_TOKEN = "0xfFf9976782d46CC05630D1f6eBAb18b2324d6B14" # WETH +SELL_TOKEN = "0xbe72E441BF55620febc26715db68d3494213D8Cb" # USDC +SELL_AMOUNT_BEFORE_FEE = "100000000000000000000" # 100 USDC with 18 decimals +ORDER_KIND = "sell" +CHAIN = Chain.SEPOLIA +CHAIN_ID = SupportedChainId.SEPOLIA + + +ORDER_BOOK_API = OrderBookApi(context={"chain_id": CHAIN_ID}) +ADDRESS = os.getenv("USER_ADDRESS") +ACCOUNT = Account.from_key(os.getenv("PRIVATE_KEY")) + + +async def get_order_quote( + order_quote_request: OrderQuoteRequest, order_side: OrderQuoteSide +) -> OrderQuoteResponse: + return await ORDER_BOOK_API.post_quote(order_quote_request, order_side) + + +def sign_order(order: Order) -> EcdsaSignature: + order_domain = asdict( + domain( + chain=CHAIN, verifying_contract=CowContractAddress.SETTLEMENT_CONTRACT.value + ) + ) + del order_domain["salt"] # TODO: improve interfaces + + return _sign_order(order_domain, order, ACCOUNT, SigningScheme.EIP712) + + +async def post_order(order: Order, signature: EcdsaSignature) -> UID: + order_creation = OrderCreation( + sellToken=order.sellToken, + buyToken=order.buyToken, + sellAmount=order.sellAmount, + feeAmount=order.feeAmount, + buyAmount=order.buyAmount, + validTo=order.validTo, + kind=order.kind, + partiallyFillable=order.partiallyFillable, + appData=order.appData, + signature=signature.data, + signingScheme="eip712", + receiver=order.receiver, + **{"from": ADDRESS}, + ) + return await ORDER_BOOK_API.post_order(order_creation) + + +async def main(): + order_quote_request = OrderQuoteRequest( + **{ + "sellToken": SELL_TOKEN, + "buyToken": BUY_TOKEN, + "from": ADDRESS, + } + ) + order_side = OrderQuoteSide( + kind=ORDER_KIND, sellAmountBeforeFee=SELL_AMOUNT_BEFORE_FEE + ) + + order_quote = await get_order_quote(order_quote_request, order_side) + + order_quote_dict = json.loads(order_quote.quote.json(by_alias=True)) + order = Order( + **{ + "sellToken": SELL_TOKEN, + "buyToken": BUY_TOKEN, + "receiver": ADDRESS, + "validTo": order_quote_dict["validTo"], + "appData": "0x0000000000000000000000000000000000000000000000000000000000000000", + "sellAmount": SELL_AMOUNT_BEFORE_FEE, # Since it is a sell order, the sellAmountBeforeFee is the same as the sellAmount + "buyAmount": order_quote_dict["buyAmount"], + "feeAmount": "0", # CoW Swap does not charge fees + "kind": ORDER_KIND, + "sellTokenBalance": "erc20", + "buyTokenBalance": "erc20", + } + ) + + signature = sign_order(order) + order_uid = await post_order(order, signature) + print(f"order posted on link: {ORDER_BOOK_API.get_order_link(order_uid.root)}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/order_book/test_api.py b/tests/order_book/test_api.py index c2ffe49..d4c8bd4 100644 --- a/tests/order_book/test_api.py +++ b/tests/order_book/test_api.py @@ -2,12 +2,13 @@ from cow_py.order_book.generated.model import ( UID, OrderQuoteResponse, + OrderQuoteSide, Trade, OrderQuoteRequest, ) import httpx from unittest.mock import AsyncMock, patch -from cow_py.order_book import ( +from cow_py.order_book.api import ( OrderBookApi, OrderCreation, ) @@ -50,18 +51,22 @@ async def test_get_trades_by_order_uid(self): self.assertEqual(trades, [mock_trade]) async def test_post_quote(self): - mock_order_quote_request_data = { - "sellToken": "0x", - "buyToken": "0x", - "receiver": "0x", - "appData": "app_data_object", - "appDataHash": "0x", - "from": "0x", - "priceQuality": "verified", - "signingScheme": "eip712", - "onchainOrder": False, - } - mock_order_quote_request = OrderQuoteRequest(**mock_order_quote_request_data) + mock_order_quote_request = OrderQuoteRequest( + **{ + "sellToken": "0x", + "buyToken": "0x", + "receiver": "0x", + "appData": "app_data_object", + "appDataHash": "0x", + "from": "0x", + "priceQuality": "verified", + "signingScheme": "eip712", + "onchainOrder": False, + } + ) + mock_order_quote_side = OrderQuoteSide( + **{"sellAmountBeforeFee": "0", "kind": "sell"} + ) mock_order_quote_response_data = { "quote": { "sellToken": "0x", @@ -87,7 +92,9 @@ async def test_post_quote(self): mock_request.return_value = httpx.Response( 200, json=mock_order_quote_response_data ) - response = await self.api.post_quote(mock_order_quote_request) + response = await self.api.post_quote( + mock_order_quote_request, mock_order_quote_side + ) mock_request.assert_called_once() self.assertEqual(response, mock_order_quote_response)