Python Wizardry: Fetching Liquidity Pools from Hive Engine with a Simple Script

So, you fancy dabbling with liquidity pools but don’t want to leave your cozy Python cocoon? Well, you're in luck! I've whipped up a small Python script — because why not add more joy to your life, right? — that fetches token pairs for any account you specify. It’s like Pokémon, but instead of Pikachu, you're catching SWAP.ETH. Here’s how you can use it:
python fetch_liquidity_pools.py, or use attributes like this:
python fetch_liquidity_pools.py accountname TOKEN(:OTHERTOKEN)

By default, this script keeps it cool with hive-engine and BTC. You can try ALL if you're feeling adventurous, but be warned — Hive Engine has more pairs than a shoe store, and the script might just throw a tantrum (hence why BTC is the default).

Example of the Magic in Action:

Run it like this:

python fetch_liquidity_pools.py hive-engine ETH

And voila, you'll see something like this:

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

Using the Script Like a Pro:

If you’re the type who loves to DIY, you can also use the script as a function in your own scripts. Just do a:

from fetch_liquidity_pools import get_filtered_pools

And Bob’s your uncle!

The Secret Sauce — AKA the Script Itself:

Now, I know you’re itching to see the magic behind the curtain, so here it is — all 99 lines of it. (Yes, I counted. Twice.) Don’t worry if it looks like hieroglyphics; I've sprinkled some comments to help you along the way.

# fetch_liquidity_pools.py
import argparse
import requests
from time import sleep

# Hive-Engine API Node
HIVE_ENGINE_NODE = 'https://api2.hive-engine.com/rpc/contracts'
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 token to filter, or use 'ALL' to list all tokens

def fetch_liquidity_positions(account_name):
    # Fetch liquidity positions for the given account
    url = HIVE_ENGINE_NODE
    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)

    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 fetch_pool_details(token_pair):
    # Fetch details of the specified liquidity pool
    for attempt in range(max_retries):
        url = HIVE_ENGINE_NODE
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "find",
            "params": {
                "contract": "marketpools",
                "table": "pools",
                "query": {"tokenPair": token_pair},
                "limit": 1
            }
        }

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

        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']:
                return data['result'][0]

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

    return {}

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

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

        # 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():
            continue

        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:
            continue

        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"

        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
        })

    return filtered_pools

def main(account_name, filter_token):
    # Main function to 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']:.6f} {pool['base_symbol']} | "
              f"Quote Balance: {pool['quote_balance']:.6f} {pool['quote_symbol']}")

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)

Give it a spin and let me know if it brings you joy or just mild frustration. Either way, I’m here for the feedback!