{ "cells": [ { "cell_type": "markdown", "id": "bound-olive", "metadata": {}, "source": [ "# ⚠ Warning\n", "\n", "THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.\n", "\n", "[![Binder](https://mybinder.org/badge_logo.svg)](https://mybinder.org/v2/gl/OpinionatedGeek%2Fmango-explorer/HEAD?filepath=Context.ipynb) _🏃‍♀️ To run this notebook press the ⏩ icon in the toolbar above._\n", "\n", "[🥭 Mango Markets](https://mango.markets/) support is available at: [Docs](https://docs.mango.markets/) | [Discord](https://discord.gg/67jySBhxrg) | [Twitter](https://twitter.com/mangomarkets) | [Github](https://github.com/blockworks-foundation) | [Email](mailto:hello@blockworks.foundation)" ] }, { "cell_type": "markdown", "id": "adjustable-fruit", "metadata": {}, "source": [ "# 🥭 Context\n", "\n", "This notebook contains a `Context` object to manage Solana connection configuration and Mango groups." ] }, { "cell_type": "markdown", "id": "interesting-madness", "metadata": {}, "source": [ "## Environment Variables\n", "\n", "It's possible to override the values in the `Context` variables provided. This can be easier than creating the `Context` in code or introducing dependencies and configuration.\n", "\n", "The following environment variables are read:\n", "* CLUSTER (defaults to: mainnet-beta)\n", "* CLUSTER_URL (defaults to URL for RPC server for CLUSTER defined in `ids.json`)\n", "* GROUP_NAME (defaults to: BTC_ETH_USDT)\n" ] }, { "cell_type": "markdown", "id": "lined-underwear", "metadata": {}, "source": [ "## Provided Configured Objects\n", "\n", "This notebook provides 3 `Context` objects, already configured and ready to use.\n", "* default_context (uses the environment variables specified above and `ids.json` file for configuration)\n", "* solana_context (uses the environment variables specified above and `ids.json` file for configuration but explicitly sets the RPC server to be [Solana's mainnet RPC server](https://api.mainnet-beta.solana.com))\n", "* serum_context (uses the environment variables specified above and `ids.json` file for configuration but explicitly sets the RPC server to be [Project Serum's mainnet RPC server](https://solana-api.projectserum.com))\n", "\n", "Where notebooks depend on `default_context`, you can change this behaviour by adding an import line like:\n", "```\n", "from Context import solana_context as default_context\n", "```\n", "This can be useful if one of the RPC servers starts behaving oddly." ] }, { "cell_type": "code", "execution_count": null, "id": "german-exhibition", "metadata": { "jupyter": { "source_hidden": true } }, "outputs": [], "source": [ "import logging\n", "import os\n", "import random\n", "import time\n", "import typing\n", "\n", "from decimal import Decimal\n", "from solana.publickey import PublicKey\n", "from solana.rpc.api import Client\n", "from solana.rpc.types import MemcmpOpts, TokenAccountOpts, RPCError, RPCMethod, RPCResponse\n", "from solana.rpc.commitment import Single\n", "from spl.token._layouts import ACCOUNT_LAYOUT\n", "\n", "from Constants import MangoConstants, SOL_DECIMAL_DIVISOR\n", "from Decoder import decode_binary, encode_binary\n" ] }, { "cell_type": "markdown", "id": "industrial-performer", "metadata": {}, "source": [ "## AccountInfo class\n", "\n", "This - unlike the classes in [BaseModel](BaseModel.ipynb) - is quite low-level and needs to be handled differently. It's here because then the `Context` object can depend on it, while putting it in [BaseModel](BaseModel.ipynb) would introduce a dependency loop." ] }, { "cell_type": "code", "execution_count": null, "id": "aerial-reggae", "metadata": {}, "outputs": [], "source": [ "class AccountInfo:\n", " def __init__(self, address: PublicKey, executable: bool, lamports: Decimal, owner: PublicKey, rent_epoch: Decimal, data: bytes):\n", " self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)\n", " self.address: PublicKey = address\n", " self.executable: bool = executable\n", " self.lamports: Decimal = lamports\n", " self.owner: PublicKey = owner\n", " self.rent_epoch: Decimal = rent_epoch\n", " self.data: bytes = data\n", "\n", " def encoded_data(self) -> typing.List:\n", " return encode_binary(self.data)\n", "\n", " def __str__(self) -> str:\n", " return f\"\"\"« AccountInfo [{self.address}]:\n", " Owner: {self.owner}\n", " Executable: {self.executable}\n", " Lamports: {self.lamports}\n", " Rent Epoch: {self.rent_epoch}\n", "»\"\"\"\n", "\n", " def __repr__(self) -> str:\n", " return f\"{self}\"\n", "\n", " @staticmethod\n", " def _from_response_values(response_values: typing.Dict[str, typing.Any], address: PublicKey) -> \"AccountInfo\":\n", " executable = bool(response_values[\"executable\"])\n", " lamports = Decimal(response_values[\"lamports\"])\n", " owner = PublicKey(response_values[\"owner\"])\n", " rent_epoch = Decimal(response_values[\"rentEpoch\"])\n", " data = decode_binary(response_values[\"data\"])\n", " return AccountInfo(address, executable, lamports, owner, rent_epoch, data)\n", "\n", " @staticmethod\n", " def from_response(response: RPCResponse, address: PublicKey) -> \"AccountInfo\":\n", " return AccountInfo._from_response_values(response[\"result\"][\"value\"], address)\n" ] }, { "cell_type": "markdown", "id": "mathematical-following", "metadata": {}, "source": [ "## Context class" ] }, { "cell_type": "code", "execution_count": null, "id": "approved-anthropology", "metadata": {}, "outputs": [], "source": [ "class Context:\n", " def __init__(self, cluster: str, cluster_url: str, program_id: PublicKey, dex_program_id: PublicKey,\n", " group_name: str, group_id: PublicKey):\n", " self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)\n", " self.cluster: str = cluster\n", " self.cluster_url: str = cluster_url\n", " self.client: Client = Client(cluster_url)\n", " self.program_id: PublicKey = program_id\n", " self.dex_program_id: PublicKey = dex_program_id\n", " self.group_name: str = group_name\n", " self.group_id: PublicKey = group_id\n", " self.commitment = Single\n", " self.encoding:str = \"base64\"\n", "\n", "\n", " def fetch_sol_balance(self, account_public_key: PublicKey) -> Decimal:\n", " result = self.client.get_balance(account_public_key, commitment=self.commitment)\n", " value = Decimal(result[\"result\"][\"value\"])\n", " return value / SOL_DECIMAL_DIVISOR\n", "\n", "\n", " def fetch_token_balance(self, account_public_key: PublicKey, token_mint: PublicKey) -> typing.Optional[Decimal]:\n", " opts = TokenAccountOpts(mint = token_mint)\n", "\n", " token_accounts_response = self.client.get_token_accounts_by_owner(account_public_key, opts, commitment=self.commitment)\n", " token_accounts = token_accounts_response[\"result\"][\"value\"]\n", " if len(token_accounts) == 0:\n", " return None\n", "\n", " total_value = Decimal(0)\n", " for token_account in token_accounts:\n", " result = self.client.get_token_account_balance(token_account[\"pubkey\"], commitment=self.commitment)\n", " value = Decimal(result[\"result\"][\"value\"][\"amount\"])\n", " decimal_places = result[\"result\"][\"value\"][\"decimals\"]\n", " divisor = Decimal(10 ** decimal_places)\n", " total_value += value / divisor\n", "\n", " return total_value\n", "\n", "\n", " def fetch_largest_token_account_for_owner(self, owner_public_key: PublicKey, token_mint: PublicKey) -> typing.Optional[AccountInfo]:\n", " opts = TokenAccountOpts(mint = token_mint)\n", "\n", " token_accounts_response = self.client.get_token_accounts_by_owner(owner_public_key, opts, commitment=self.commitment)\n", "\n", " largest_amount = -1\n", " largest_account: typing.Optional[AccountInfo] = None\n", " for token_account_response in token_accounts_response[\"result\"][\"value\"]:\n", " data = decode_binary(token_account_response[\"account\"][\"data\"])\n", " account_data = ACCOUNT_LAYOUT.parse(data)\n", " if account_data.amount > largest_amount:\n", " largest_amount = account_data.amount\n", " largest_account = AccountInfo(PublicKey(token_account_response[\"pubkey\"]),\n", " token_account_response[\"account\"][\"executable\"],\n", " Decimal(token_account_response[\"account\"][\"lamports\"]),\n", " PublicKey(token_account_response[\"account\"][\"owner\"]),\n", " Decimal(token_account_response[\"account\"][\"rentEpoch\"]),\n", " data)\n", "\n", " return largest_account\n", "\n", "\n", " def load_account(self, address: PublicKey) -> typing.Optional[\"AccountInfo\"]:\n", " response: RPCResponse = self.client.get_account_info(address)\n", " self.unwrap_transaction_id_or_raise_exception(response)\n", " if response[\"result\"][\"value\"] is None:\n", " return None\n", "\n", " return AccountInfo.from_response(response, address)\n", "\n", "\n", " def load_multiple_accounts(self, addresses: typing.List[PublicKey]) -> typing.List[\"AccountInfo\"]:\n", " address_strings = list(map(PublicKey.__str__, addresses))\n", " response = self.client._provider.make_request(RPCMethod(\"getMultipleAccounts\"), address_strings)\n", " response_value_list = zip(response[\"result\"][\"value\"], addresses)\n", " return list(map(lambda pair: AccountInfo._from_response_values(pair[0], pair[1]), response_value_list))\n", "\n", "\n", " def fetch_program_accounts_for_owner(self, program_id: PublicKey, owner: PublicKey):\n", " memcmp_opts = [\n", " MemcmpOpts(offset=40, bytes=str(owner)),\n", " ]\n", "\n", " return self.client.get_program_accounts(program_id, memcmp_opts=memcmp_opts, commitment=self.commitment, encoding=self.encoding)\n", "\n", "\n", " def unwrap_transaction_id_or_raise_exception(self, response: RPCResponse) -> str:\n", " if \"error\" in response:\n", " if response[\"error\"] is str:\n", " message: str = typing.cast(str, response[\"error\"])\n", " code: int = -1\n", " else:\n", " error: RPCError = typing.cast(RPCError, response[\"error\"])\n", " message = error[\"message\"]\n", " code = error[\"code\"]\n", " raise Exception(f\"Error response from server: '{message}', code: {code}\")\n", "\n", " return response[\"result\"]\n", "\n", "\n", " def random_client_id(self) -> int:\n", " # 9223372036854775807 is sys.maxsize for 64-bit systems, with a bit_length of 63.\n", " # We explicitly want to use a max of 64-bits though, so we use the number instead of\n", " # sys.maxsize, which could be lower on 32-bit systems or higher on 128-bit systems.\n", " return random.randrange(9223372036854775807)\n", "\n", "\n", " @staticmethod\n", " def _lookup_name_by_address(address: PublicKey, collection: typing.Dict[str, str]) -> typing.Optional[str]:\n", " address_string = str(address)\n", " for stored_name, stored_address in collection.items():\n", " if address_string == stored_address:\n", " return stored_name\n", " return None\n", "\n", " def lookup_market_name(self, market_address: PublicKey) -> str:\n", " return Context._lookup_name_by_address(market_address, MangoConstants[self.cluster][\"spot_markets\"]) or \"« Unknown Market »\"\n", "\n", " def lookup_oracle_name(self, token_address: PublicKey) -> str:\n", " return Context._lookup_name_by_address(token_address, MangoConstants[self.cluster][\"oracles\"]) or \"« Unknown Oracle »\"\n", "\n", " def lookup_token_name(self, token_address: PublicKey) -> str:\n", " return Context._lookup_name_by_address(token_address, MangoConstants[self.cluster][\"symbols\"]) or \"« Unknown Token »\"\n", "\n", " def wait_for_confirmation(self, transaction_id: str, max_wait_in_seconds: int = 60) -> None:\n", " self.logger.info(f\"Waiting up to {max_wait_in_seconds} seconds for {transaction_id}.\")\n", " for wait in range(0, max_wait_in_seconds):\n", " time.sleep(1)\n", " confirmed = default_context.client.get_confirmed_transaction(transaction_id)\n", " if confirmed[\"result\"] is not None:\n", " self.logger.info(f\"Confirmed after {wait} seconds.\")\n", " return\n", " self.logger.info(f\"Timed out after {wait} seconds waiting on transaction {transaction_id}.\")\n", "\n", " def __str__(self) -> str:\n", " return f\"\"\"« Context:\n", " Cluster: {self.cluster}\n", " Cluster URL: {self.cluster_url}\n", " Program ID: {self.program_id}\n", " DEX Program ID: {self.dex_program_id}\n", " Group Name: {self.group_name}\n", " Group ID: {self.group_id}\n", "»\"\"\"\n", "\n", " def __repr__(self) -> str:\n", " return f\"{self}\"\n" ] }, { "cell_type": "markdown", "id": "sufficient-portrait", "metadata": {}, "source": [ "## default_context object\n", "\n", "A default `Context` object that connects to mainnet, to save having to create one all over the place. This `Context` uses the default values in the `ids.json` file, overridden by environment variables if they're set." ] }, { "cell_type": "code", "execution_count": null, "id": "cellular-chest", "metadata": {}, "outputs": [], "source": [ "default_cluster = os.environ.get(\"CLUSTER\") or \"mainnet-beta\"\n", "default_cluster_url = os.environ.get(\"CLUSTER_URL\") or MangoConstants[\"cluster_urls\"][default_cluster]\n", "\n", "default_program_id = PublicKey(MangoConstants[default_cluster][\"mango_program_id\"])\n", "default_dex_program_id = PublicKey(MangoConstants[default_cluster][\"dex_program_id\"])\n", "\n", "default_group_name = os.environ.get(\"GROUP_NAME\") or \"BTC_ETH_USDT\"\n", "default_group_id = PublicKey(MangoConstants[default_cluster][\"mango_groups\"][default_group_name][\"mango_group_pk\"])\n", "\n", "default_context = Context(default_cluster, default_cluster_url, default_program_id,\n", " default_dex_program_id, default_group_name, default_group_id)" ] }, { "cell_type": "markdown", "id": "renewable-jerusalem", "metadata": {}, "source": [ "## solana_context object\n", "\n", "A `Context` object that connects to mainnet using Solana's own https://api.mainnet-beta.solana.com server. Apart from the RPC server URL, this `Context` uses the default values in the `ids.json` file, overridden by environment variables if they're set." ] }, { "cell_type": "code", "execution_count": null, "id": "deluxe-immigration", "metadata": {}, "outputs": [], "source": [ "solana_cluster_url = \"https://api.mainnet-beta.solana.com\"\n", "\n", "solana_context = Context(default_cluster, solana_cluster_url, default_program_id,\n", " default_dex_program_id, default_group_name, default_group_id)" ] }, { "cell_type": "markdown", "id": "celtic-concentrate", "metadata": {}, "source": [ "## serum_context object\n", "\n", "A `Context` object that connects to mainnet using Serum's own https://solana-api.projectserum.com server. Apart from the RPC server URL, this `Context` uses the default values in the `ids.json` file, overridden by environment variables if they're set." ] }, { "cell_type": "code", "execution_count": null, "id": "careful-asthma", "metadata": {}, "outputs": [], "source": [ "serum_cluster_url = \"https://solana-api.projectserum.com\"\n", "\n", "serum_context = Context(default_cluster, serum_cluster_url, default_program_id,\n", " default_dex_program_id, default_group_name, default_group_id)" ] }, { "cell_type": "markdown", "id": "aware-realtor", "metadata": {}, "source": [ "# 🏃 Running\n", "\n", "If running interactively, just print out the default Context object." ] }, { "cell_type": "code", "execution_count": null, "id": "reflected-kentucky", "metadata": {}, "outputs": [], "source": [ "if __name__ == \"__main__\":\n", " logging.getLogger().setLevel(logging.INFO)\n", "\n", " print(default_context)\n", "\n", " single_account_info = default_context.load_account(default_context.dex_program_id)\n", " print(\"DEX account info\", single_account_info)\n", "\n", " multiple_account_info = default_context.load_multiple_accounts([default_context.program_id, default_context.dex_program_id])\n", " print(\"Mango program and DEX account info\", multiple_account_info)\n", "\n", " print(\"Lookup ETH token name result:\", default_context.lookup_token_name(PublicKey(\"2FPyTwcZLUg1MDrwsyoP4D6s1tM7hAkHYRjkNb5w6Pxk\")))\n", " print(\"Lookup BTC/USDC market name result:\", default_context.lookup_market_name(PublicKey(\"CVfYa8RGXnuDBeGmniCcdkBwoLqVxh92xB1JqgRQx3F\")))\n", "\n", " # Fill out your account address between the quotes below\n", " MY_ACCOUNT_ADDRESS = \"\"\n", " # Don't edit anything beyond here.\n", "\n", " if MY_ACCOUNT_ADDRESS != \"\":\n", " account_key = PublicKey(MY_ACCOUNT_ADDRESS)\n", " print(\"SOL balance:\", default_context.fetch_sol_balance(account_key))\n", " print(\"SRM balance:\", default_context.fetch_token_balance(account_key, PublicKey(\"SRMuApVNdxXokk5GT7XD5cUUgXMBCoAz2LHeuAoKWRt\")))\n", " print(\"SRM AccountInfo:\", default_context.fetch_largest_token_account_for_owner(account_key, PublicKey(\"SRMuApVNdxXokk5GT7XD5cUUgXMBCoAz2LHeuAoKWRt\")))\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.8.6" }, "toc": { "base_numbering": 1, "nav_menu": {}, "number_sections": true, "sideBar": true, "skip_h1_title": false, "title_cell": "Table of Contents", "title_sidebar": "Contents", "toc_cell": false, "toc_position": {}, "toc_section_display": true, "toc_window_display": true }, "varInspector": { "cols": { "lenName": 16, "lenType": 16, "lenVar": 40 }, "kernels_config": { "python": { "delete_cmd_postfix": "", "delete_cmd_prefix": "del ", "library": "var_list.py", "varRefreshCmd": "print(var_dic_list())" }, "r": { "delete_cmd_postfix": ") ", "delete_cmd_prefix": "rm(", "library": "var_list.r", "varRefreshCmd": "cat(var_dic_list()) " } }, "types_to_exclude": [ "module", "function", "builtin_function_or_method", "instance", "_Feature" ], "window_display": false } }, "nbformat": 4, "nbformat_minor": 5 }