feat: improve test stability by optional pricing source (#490)

* feat: improve test stability by optional pricing source

* add helper function to process each pool

tests/README.md

@@ -107,6 +107,7 @@ There are 3 possible triggers:
 Our `pytest` parses the following environment variables in `conftest.py`
 
 - `SQS_API_KEY` -> API Key to bypass rate limit. If not provided, the tests will run without API key set.
+- `COINGECKO_API_KEY` -> API key to Coingecko pricing service. If specified, token price will be fetched from Coingecko when calculating pool liquidity capitalization value during test executions
 - `SQS_ENVIRONMENTS` -> Comma separated list of environment names per "Supported Environments" to run the tests against. If not provided, the tests will run against stage.
 
 ## Geo-Distributed Synthetic Monitoring

tests/asset_list_service.py

@@ -0,0 +1,28 @@
+import requests
+
+ASSET_LIST_URL = "https://raw.githubusercontent.com/osmosis-labs/assetlists/main/osmosis-1/generated/frontend/assetlist.json"
+
+class AssetListService:
+    # cache the asset list, key => coinMinimalDenom, value => {symbol, decimals, coingeckoId}
+    asset_map = {}
+
+    # This is a simple service that fetches the asset list from the asset list URL
+    def get_asset_metadata(self, denom):
+        if self.asset_map == {}:    
+            response = requests.get(ASSET_LIST_URL)
+            if response.status_code != 200:
+                raise Exception(f"Error fetching asset list: {response.text}")
+            asset_list = response.json().get("assets", [])
+            for asset in asset_list:
+                coinMinimalDenom = asset.get("coinMinimalDenom")
+                symbol = asset.get("symbol")
+                decimals = asset.get("decimals")
+                coingeckoId = asset.get("coingeckoId")
+                self.asset_map[coinMinimalDenom] = {
+                    "symbol": symbol,
+                    "decimals": decimals,
+                    "coingeckoId": coingeckoId
+                }
+        return self.asset_map.get(denom, None)
+
+

tests/coingecko_service.py

@@ -1,11 +1,23 @@
 import requests
+import os
 
 COINGECKO_URL = "https://prices.osmosis.zone/api/v3/simple/price"
 USD_CURRENCY = "usd"
 
 class CoingeckoService:
+
+    # Caching token price is still acceptable for test purposes 
+    # since the token price is not expected to change a lot during test executions
+    # key => coingecko_id, value => token price
     cache = {}
 
+    def __init__(self, coingecko_api_key):
+        self.coingecko_api_key = coingecko_api_key
+
+    # Check if the service is available
+    def isServiceAvailable(self):
+        return self.coingecko_api_key is not None
+
     # Given the coingecko id, call the coingecko API endpoint and return its token price
     def get_token_price(self, coingecko_id):
         if coingecko_id in self.cache:
@@ -16,8 +28,21 @@ def get_token_price(self, coingecko_id):
             "ids": coingecko_id,
             "vs_currencies": USD_CURRENCY
         }
+
         # Send the GET request
-        response = requests.get(COINGECKO_URL, params=params)
+        # Set the auth header if the API key is provided
+        # If not provided, the request will be sent without the auth header 
+        # but expect a lower rate limit and getting HTTP 429 error
+        if self.coingecko_api_key is not None:
+            headers = {
+                "X-API-KEY": self.coingecko_api_key
+            }
+            response = requests.get(COINGECKO_URL, params=params, headers=headers)
+        else:
+            response = requests.get(COINGECKO_URL, params=params)
+
+        if response.status_code == 429:
+            raise Exception(f"Too many requests to {COINGECKO_URL}: {response.text}")
         if response.status_code != 200:
             raise Exception(f"Error fetching price from coingecko: {response.text}")
 

tests/conftest.py

@@ -15,6 +15,7 @@
 from chain_service import ChainService
 from util import *
 from decimal import *
+from asset_list_service import AssetListService
 
 
 
@@ -31,10 +32,23 @@ def parse_api_key():
 
 api_key = parse_api_key()
 
+def parse_coingecko_api_key():
+    """
+    Parse the COINGECKO_API_KEY environment variable and return it
+
+    If the environment variable is not set, the default API key is ""
+    """
+    coingecko_api_key = os.getenv('COINGECKO_API_KEY', None)
+
+    return coingecko_api_key
+
+coingecko_api_key = parse_coingecko_api_key()
+
 SERVICE_SQS_STAGE = SQSService(SQS_STAGE, api_key)
 SERVICE_SQS_PROD = SQSService(SQS_PROD, api_key)
 SERVICE_SQS_LOCAL = SQSService(SQS_LOCAL, api_key)
-SERVICE_COINGECKO = CoingeckoService()
+SERVICE_COINGECKO = CoingeckoService(coingecko_api_key)
+SERVICE_ASSET_LIST = AssetListService()
 
 STAGE_INPUT_NAME = "stage"
 PROD_INPUT_NAME = "prod"
@@ -557,6 +571,11 @@ def pytest_sessionstart(session):
     """
     print("Session is starting. Worker ID:", getattr(session.config, 'workerinput', {}).get('workerid', 'master'))
 
+    if conftest.SERVICE_COINGECKO.isServiceAvailable():
+        print("Using Coingecko to calculate pool liquidity capitalization")
+    else:
+        print("Using Numia to calculate pool liquidity capitalization")
+
     global shared_test_state
 
     # Example setup logic

tests/data_service.py

@@ -1,4 +1,7 @@
 import requests
+from decimal import Decimal
+
+import conftest
 
 # Endpoint URLs
 NUMIA_API_URL = 'https://stage-proxy-data-api.osmosis-labs.workers.dev'
@@ -16,6 +19,41 @@ def fetch_tokens():
     except requests.exceptions.RequestException as e:
         print(f"Error fetching data from Numia: {e}")
         return []
+
+# Helper function to calcualte token cap given the token amount 
+# and the token price from coingecko
+def get_token_cap(denom, amount_str):
+    token_metadata = conftest.SERVICE_ASSET_LIST.get_asset_metadata(denom)
+    if token_metadata is None:
+        return None
+    if token_metadata.get('coingeckoId') is None:
+        return None
+    token_price = conftest.SERVICE_COINGECKO.get_token_price(token_metadata.get('coingeckoId'))
+    if token_price is None:
+        return None
+    return Decimal(amount_str) * Decimal(token_price) 
+
+# Helper function to update the pool liquidity cap using Coingecko
+# abort the update if any token cap is not available
+def update_pool_liquidity_cap(pool):
+    try:
+        total_pool_cap = Decimal(0)
+        all_token_cap_captured = True # flag to check if all token cap is captured, if not, no update will be made
+        tokens = pool.get('pool_tokens', []) 
+        if not isinstance(tokens, list): # tokens from numia can be a dict or a list
+            tokens = tokens.values()
+        for token in tokens:
+            denom = token.get('denom')
+            token_cap = get_token_cap(denom, token['amount'])
+            if token_cap is None:
+                all_token_cap_captured = False
+                break
+            total_pool_cap += token_cap
+        if all_token_cap_captured:
+            pool.update({'liquidity': float(total_pool_cap)})
+    except Exception as e:
+        print(f"warning: error processing pool data (pool id {pool.get('pool_id')}) using Coingecko: {e}. fallback to numia data")
+        return
 
 def fetch_pools():
     """Fetches all pools by iterating through paginated results."""
@@ -33,6 +71,13 @@ def fetch_pools():
             pools = data.get('pools', [])
             pagination = data.get('pagination', {})
 
+            # Calculate the pool liquidity cap using Coingecko if service is available, aka the API key is provided thru env var
+            if conftest.SERVICE_COINGECKO.isServiceAvailable():
+                # Iterate through the pools and update the **existing** pool liquidity cap data with a more accurate value from Coingecko
+                for pool in pools:
+                    for pool in pools:
+                        update_pool_liquidity_cap(pool)
+
             # Add this batch to the accumulated pool data
             all_pools.extend(pools)