Fetching Hive-Engine Liquidity Pools Just Got Better!

(edited)

Remember that Python script I shared with you all a couple of days ago? The one that fetches liquidity pools and makes you feel like a wizard commanding the blockchain seas? Well, I've been tinkering, and now it's even better!

What's New?

Redwing
An unrelated photo of a Redwing,
shot by myself.

The updated version of the script now includes a JSON file that holds a list of Hive-Engine nodes to cycle through, making the script more resilient. The idea was to have a robust list of nodes, but... well, let's just say I hit a few bumps along the road. Currently, we only have two functioning hive-engine.com servers on the list.

But hey, two is better than none, right? ๐Ÿ˜…

Running the Improved Script

You can still run the script like before:

python fetch_liquidity_pools.py accountname TOKEN(:OTHERTOKEN)

Or if you're just feeling a bit more specific:

python fetch_liquidity_pools.py hive-engine ETH

The Enhanced Features:

  • Node Switching: The script now loads a list of nodes from a JSON file (nodes.json) and cycles through them, selecting one randomly to handle your request. This helps ensure that if one node is down or slow, another can step in.
  • Caching Token Details: We now store token details in a local cache (token_details_cache.json). This prevents repeatedly fetching the same data, saving time and network bandwidth.
  • More Error Handling: Because we all know that dealing with APIs can be like herding cats, more robust error handling has been added. Now, you'll get better feedback on what went wrong, with retry mechanisms to recover from temporary issues. (You can uncomment the debugging statements to see a whole lot more what's happening behind the scenes!)

Example Output

To remind you of the magic, here's what the output looks like when fetching positions with the ETH token:

Liquidity Pool Positions with ETH token:
Token Pair: SWAP.HIVE:SWAP.ETH | Base Balance: 31206.634533 SWAP.HIVE | Quote Balance: 2.242819 SWAP.ETH
Token Pair: BEE:SWAP.ETH | Base Balance: 42215.817136 BEE | Quote Balance: 1.422489 SWAP.ETH
Token Pair: SWAP.BTC:SWAP.ETH | Base Balance: 0.007309 SWAP.BTC | Quote Balance: 0.169925 SWAP.ETH
Token Pair: SWAP.ETH:SWAP.USDT | Base Balance: 0.151195 SWAP.ETH | Quote Balance: 351.954867 SWAP.USDT
Token Pair: SWAP.ETH:SPS | Base Balance: 0.033753 SWAP.ETH | Quote Balance: 11164.970135 SPS

As easily, you can call it from your own Python scripts by importing it...

Using the script like a Pro

If you love incorporating ready-made tools (or just can't resist tinkering), you can do that too!

Here's how:

Just add this line to your script to access to the functions:

from fetch_liquidity_pools import get_filtered_pools

Once youโ€™ve imported the script, you can call the get_filtered_pools function with your own parameters:

# Import the necessary function from your script
from fetch_liquidity_pools import get_filtered_pools

def sort_pools_by_balance(account_name, filter_token):
    # Fetch the liquidity pools
    pools = get_filtered_pools(account_name, filter_token)
    
    # Sort the pools by base balance in descending order
    sorted_pools = sorted(pools, key=lambda x: x['base_balance'], reverse=True)
    
    print(f"\n--- Sorted Liquidity Pools for {filter_token.upper()} ---")
    for pool in sorted_pools:
        print(f"Token Pair: {pool['token_pair']} | Base Balance: {pool['base_balance']:.{pool['base_precision']}f} {pool['base_symbol']} | "
              f"Quote Balance: {pool['quote_balance']:.{pool['quote_precision']}f} {pool['quote_symbol']}")

if __name__ == "__main__":
    # Example usage with your Hive account and desired token filter
    account_name = 'your-hive-account'
    filter_token = 'ETH'  # Replace with the token you want to filter by

    sort_pools_by_balance(account_name, filter_token)

This script is just an example, it might or might not work as advertised.

Why Import?

By importing the script into your own programs, you can:

  • Sort the pools (see the example)
  • Automate repetitive tasks by fetching and filtering liquidity pools programmatically.
  • Integrate liquidity data into larger trading or analysis pipelines.
  • Customize the logic further to suit your specific needs.

๐Ÿ’กA Few Tips:

Make sure the fetch_liquidity_pools.py script is in the same directory as your script or in a directory thatโ€™s part of your Python path.
Donโ€™t forget to have the nodes.json and token_details_cache.json files in the same directory as well, so the script can find them!

These are the contents of the nodes.json file:

[
  "https://api2.hive-engine.com/rpc/contracts",
  "https://api.hive-engine.com/rpc/contracts"
]

The Updated Script

Now, let's get to the juicy bits. Here's the improved script with all the bells and whistles:

# fetch_liquidity_pools.py
import json
import os
import argparse
import requests
from time import sleep
from random import choice

# Hive-Engine API Nodes
NODES_FILE = 'nodes.json'
retry_delay = 5  # seconds to wait between retries
max_retries = 3  # Maximum number of retries

# Default values
DEFAULT_ACCOUNT_NAME = 'hive-engine'  # Replace with your actual Hive account name
DEFAULT_FILTER_TOKEN = 'BTC'  # Replace with the desired default token to filter, or use 'ALL' to list all tokens

# File to store token details with precision
TOKEN_CACHE_FILE = 'token_details_cache.json'
cached_token_details = {}
hive_engine_nodes = []

def load_nodes():
    global hive_engine_nodes
    # Check if the nodes file exists
    if os.path.exists(NODES_FILE):
        try:
            with open(NODES_FILE, 'r') as f:
                hive_engine_nodes = json.load(f)
                print("Loaded Hive-Engine nodes from file.")
        except (ValueError, IOError):
            print("Error: Hive-Engine nodes file is corrupted or not readable. Please re-create 'nodes.json' with the list of nodes.")
    else:
        print("Error: Hive-Engine nodes file not found. Please create 'nodes.json' with the list of nodes.")
        hive_engine_nodes = [] # Ensure nodes list is empty on error


def get_node():
# Choose a random node from the list
    if hive_engine_nodes:
        selected_node = choice(hive_engine_nodes)
        print(f"Using Hive-Engine node: {selected_node}") # Print the current node
        return selected_node
    else:
        print("Error: No Hive-Engine nodes available.")
        return None


def load_token_cache():
    global cached_token_details
    # Check if the token cache file exists
    if os.path.exists(TOKEN_CACHE_FILE):
        try:
            with open(TOKEN_CACHE_FILE, 'r') as f:
                cached_token_details = json.load(f)
                print("Loaded cached token details from file.")
        except (ValueError, IOError):
            print("Error: Failed to load token cache file. Starting with an empty cache.")


def save_token_cache():
    # Save the current token details cache to a file
    try:
        with open(TOKEN_CACHE_FILE, 'w') as f:
            json.dump(cached_token_details, f)
            print("Saved token details to cache file.")
    except IOError:
        print("Error: Failed to save token cache file.")


def fetch_token_details(symbol):
    # Check if token details are already cached
    if symbol in cached_token_details:
        print(f"Token details for {symbol} found in cache.")
        return cached_token_details[symbol]

    print (f"Fetching token details for {symbol}...")
    # Fetch token details for the given symbol
    for attempt in range(max_retries):
        url = get_node()
        if not url:
            return {}

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "find",
            "params": {
                "contract": "tokens",
                "table": "tokens",
                "query": {"symbol": symbol},
                "limit": 1
            }
        }

        response = requests.post(url, json=payload)

        print(f"Attempt {attempt+1}: Status Code: {response.status_code}, Response: {response.text}")

        if response.status_code == 200:
            try:
                data = response.json()
            except ValueError:
                print("Error: Failed to parse JSON response.")
                return {}

            if 'result' in data and data['result']:
                cached_token_details[symbol] = data['result'][0]  # Cache the token details
                save_token_cache()  # Save cache after updating
                return data['result'][0]

        print(f"Error: Failed to fetch token details for {symbol}. Status Code: {response.status_code}")
        if attempt < max_retries - 1:
            sleep(retry_delay)
        else:
            print(f"Max retries exceeded for {symbol}. Skipping.")

    return {}


def fetch_pool_details(token_pair):
    # Fetch details of the specified liquidity pool
    for attempt in range(max_retries):
        url = get_node()
        if not url:
            print("Error: No node URL available, exiting fetch_pool_details.")
            return {}

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "find",
            "params": {
                "contract": "marketpools",
                "table": "pools",
                "query": {"tokenPair": token_pair},
                "limit": 1
            }
        }

        # print(f"Attempt {attempt + 1} to fetch pool details for {token_pair} from {url}...") # Debugging statement

        try:
            response = requests.post(url, json=payload, timeout=10) # Set a timeout for the request
            print(f"Received response status code: {response.status_code} for {token_pair} from {url}")

            if response.status_code == 200:
                try:
                    data = response.json()
                    # print(f"Data received for {token_pair}: {data}") # Debugging the received data
                    if 'result' in data and data['result']:
                        print(f"Successfully fetched pool details for {token_pair}")
                        return data['result'][0]
                    else:
                        print(f"Unexpected response format or empty result for {token_pair} from {url}: {data}")
                except ValueError as e:
                    print("Error: Failed to parse JSON response.")
                    print(f"Response content: {response.text}") # Print the actual response content
            else:
                print(f"Error: Failed to fetch pool details for {token_pair}. Status Code: {response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"Request exception occurred for {token_pair} from {url}: {e}")

        # Handle retries
        if attempt < max_retries - 1:
            print(f"Retrying after {retry_delay} seconds...")
            sleep(retry_delay)
        else:
            print(f"Max retries exceeded for {token_pair}. Skipping to next.")

    print(f"Returning empty details for {token_pair} after all attempts.")
    return {}


def fetch_liquidity_positions(account_name):
    # Fetch liquidity positions for the given account
    url = get_node()
    if not url:
        return {}

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "find",
        "params": {
            "contract": "marketpools",
            "table": "liquidityPositions",
            "query": {"account": account_name},
            "limit": 1000
        }
    }

    response = requests.post(url, json=payload)

    # print("Response Status Code: ", response.status_code)
    # print("Response Content: ", response.text) # Debug

    if response.status_code != 200:
        print(f"Error: Failed to fetch data. Status Code: {response.status_code}")
        return []

    try:
        data = response.json()
    except ValueError:
        print("Error: Failed to parse JSON response.")
        return []

    return data.get('result', [])


def get_filtered_pools(account_name, filter_token):
    # Get and filter pools by the specified token
    positions = fetch_liquidity_positions(account_name)

    # Debug: Check fetched positions
    print(f"Fetched {len(positions)} liquidity positions for account {account_name}.")

    if not positions:
        print("No liquidity positions found for this account.")
        return []

    filtered_pools = []

    for position in positions:
        token_pair = position.get('tokenPair', 'Unknown')

        # Debug: Print each position being processed
        print(f"Processing position: {position}")

        # If filter_token is 'ALL', skip filtering; otherwise, check for the token in the pair
        if filter_token.upper() != 'ALL' and filter_token.upper() not in token_pair.upper():
            print(f"Skipping position {token_pair} as it does not match filter token {filter_token.upper()}")
            continue

        # Additional debug to see which positions pass the filter
        print(f"Including position {token_pair} with filter token {filter_token.upper()}")

        # Fetch balances and calculate user share
        shares = float(position.get('shares', '0'))
        pool_details = fetch_pool_details(token_pair)
        if not pool_details:
            continue

        total_shares = float(pool_details.get('totalShares', '0'))
        base_quantity = float(pool_details.get('baseQuantity', '0'))
        quote_quantity = float(pool_details.get('quoteQuantity', '0'))

        if total_shares == 0:
            print(f"Skipping position {token_pair} due to total shares being 0.")
            continue

        # Calculate user balances
        user_base_balance = (shares / total_shares) * base_quantity
        user_quote_balance = (shares / total_shares) * quote_quantity

        if ':' in token_pair:
            base_symbol, quote_symbol = token_pair.split(':')
        else:
            base_symbol, quote_symbol = "Unknown", "Unknown"

        # Fetch token details to get precision
        base_token_details = fetch_token_details(base_symbol)
        quote_token_details = fetch_token_details(quote_symbol)
        base_precision = base_token_details.get('precision', 0)
        quote_precision = quote_token_details.get('precision', 0)

        filtered_pools.append({
            "token_pair": token_pair,
            "base_symbol": base_symbol,
            "quote_symbol": quote_symbol,
            "base_balance": user_base_balance,
            "quote_balance": user_quote_balance,
            "base_precision": base_precision,
            "quote_precision": quote_precision
        })

    # Debug: Print the number of filtered pools
    # print(f"Number of filtered pools: {len(filtered_pools)}")

    return filtered_pools


def main(account_name, filter_token):
    # Load nodes from the external file
    load_nodes()

    # Load cached token details
    load_token_cache()

    # Fetch and print filtered pools
    pools = get_filtered_pools(account_name, filter_token)
    print(f"\nLiquidity Pool Positions with {filter_token.upper()} token:")
    for pool in pools:
        print(f"Token Pair: {pool['token_pair']} | Base Balance: {pool['base_balance']:.{pool['base_precision']}f} {pool['base_symbol']} | "
              f"Quote Balance: {pool['quote_balance']:.{pool['quote_precision']}f} {pool['quote_symbol']}")

    # Debug: If no pools were printed
    # if not pools:
    #    print("No matching liquidity pools found for the given filter.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Fetch Hive-Engine liquidity pools.')
    parser.add_argument('account_name', nargs='?', default=DEFAULT_ACCOUNT_NAME, help='Hive account name to fetch liquidity pools for.')
    parser.add_argument('filter_token', nargs='?', default=DEFAULT_FILTER_TOKEN, help="Token to filter by, or 'ALL' to list all tokens.")

    args = parser.parse_args()

    main(args.account_name, args.filter_token)

What's Next?

I'd love to hear your thoughts on this update! Is it working for you, or is it just causing new headaches? Any nodes to add to the list or tips on getting those stubborn servers to talk? Let me know in the comments!

Keep experimenting, and may your code always run without errors! Cheers!

0.02854960 BEE
4 comments

Awesome! I was trying to find the time to analyze your previous script and now I have not one, but two codes to study ๐Ÿ˜… I know that this one is an improved version of the previous one, but as a beginner I have a feeling that, perhaps, it would be better if I start from the "easier" version and, once I've understood it, challange myself with this newer one :)

In any case, thanks for sharing this script! I'm slowly trying to learn Python and scripts like this one helps me a lot, as they enable me to test real world applications while I (try to) learn how they work :)

0.00540769 BEE
(edited)

Thanks so much for your kind words! ๐Ÿ˜Š I'm glad to hear that my scripts are helping you on your Python journey. I totally understand your plan to start with the 'easier' version first โ€” that's a great approach! Feel free to ask if you have any questions along the way; I'm happy to help, even if I'm a bit tired myself after battling with these for weeks. ๐Ÿ˜… (This one hasn't been the only one in the works right now, so.)

Keep pushing forward; learning by experimenting with real-world applications is one of the best ways to grasp coding concepts. Best of luck, and donโ€™t hesitate to share any thoughts or feedback! ๐Ÿš€

0E-8 BEE

Right I'm trying to finish a script I'm working since... well, I even forget when I started working on it ๐Ÿคฃ

But as soon as I'll finish it, I'm going to take your code, read it - let's be honest - a lot of times ๐Ÿ˜… and try to understand how everything works, because I've seen a lot of interesting stuff I'd love to learn!

And then I may try to use it somehow, adding my ugly personalized funcs and, lastly, probably tagging you in a post to show you how badly I used your code ๐Ÿคฃ but at least you will see that I really used it and appreciated your work :)

0.00544497 BEE

Ok, this one is definitely tougher and will require much more effort ๐Ÿ˜… I'm not going to ask you some questions now because they would be too many... let's see if I can solve some of my doubts on my own first :)

!PIZZA

0.00444375 BEE

I'm positive you will have no issues understanding it! It's littered with commented-out and non-commented debugging print statements, because I had some troubles with repeatedly calling it from another script. I eventually found a way to make it work. But it's not completely out of the woods quite yet, as I found some bugs I mentioned in that earlier comment.

Cheers!

!BEER

0E-8 BEE

It was tougher, but after getting what the fetch_token_details function was doing (at first I was like: "before there were 2 requests, now there are 3... why? Help!!" ๐Ÿคฃ), I finally succeded in understading it!

Oc I didn't see any bugs - it's already enough if I understand how it works ahaha - but there's a question I'd like to ask you, if I can:

  • how did you built the call? For example, this one:
payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "find",
            "params": {
                "contract": "tokens",
                "table": "tokens",
                "query": {"symbol": symbol},
                "limit": 1
            }
        }

Is there some documentation in the Hive API I can read to find it out? Because untill now I thought one had always to use something like the block_api, or condenser_api, or something similar, but here I can't see them!

0.00431687 BEE

Hey @arc7icwolf, here is a little bit of BEER from @gamer00 for you. Enjoy it!

We love your support by voting @detlev.witness on HIVE .

0E-8 BEE

Thanks for your contribution to the STEMsocial community. Feel free to join us on discord to get to know the rest of us!

Please consider delegating to the @stemsocial account (85% of the curation rewards are returned).

You may also include @stemsocial as a beneficiary of the rewards of this post to get a stronger support.ย 
ย 

0.00217978 BEE
(edited)

PIZZA!

$PIZZA slices delivered:
gamer00 tipped arc7icwolf
@arc7icwolf(1/10) tipped @gamer00

0E-8 BEE