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).
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
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!
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!