151 lines
5.4 KiB
Python
151 lines
5.4 KiB
Python
from hyperliquid.info import Info
|
|
from hyperliquid.utils import constants
|
|
import time
|
|
import os
|
|
import sys
|
|
|
|
def get_asset_prices(asset_names=["BTC", "ETH", "SOL", "BNB", "FARTCOIN", "PUMP", "TRUMP", "ZEC"]):
|
|
"""
|
|
Connects to the Hyperliquid API to get the current mark price of specified assets.
|
|
|
|
Args:
|
|
asset_names (list): A list of asset names to retrieve prices for.
|
|
|
|
Returns:
|
|
list: A list of dictionaries, where each dictionary contains the name and mark price of an asset.
|
|
Returns an empty list if the API call fails or no assets are found.
|
|
"""
|
|
try:
|
|
info = Info(constants.MAINNET_API_URL, skip_ws=True)
|
|
meta, asset_contexts = info.meta_and_asset_ctxs()
|
|
|
|
universe = meta.get("universe", [])
|
|
asset_data = []
|
|
|
|
for name in asset_names:
|
|
try:
|
|
index = next(i for i, asset in enumerate(universe) if asset["name"] == name)
|
|
context = asset_contexts[index]
|
|
asset_data.append({
|
|
"name": name,
|
|
"mark_price": context.get("markPx")
|
|
})
|
|
except StopIteration:
|
|
print(f"Warning: Could not find asset '{name}' in the API response.")
|
|
|
|
return asset_data
|
|
|
|
except KeyError:
|
|
print("Error: A KeyError occurred. The structure of the API response may have changed.")
|
|
return []
|
|
except Exception as e:
|
|
print(f"An unexpected error occurred: {e}")
|
|
return []
|
|
|
|
def clear_console():
|
|
# Cross-platform clear screen
|
|
if os.name == 'nt':
|
|
os.system('cls')
|
|
else:
|
|
print('\033c', end='')
|
|
|
|
def display_prices_table(prices, previous_prices):
|
|
"""
|
|
Displays a list of asset prices in a formatted table with price change indicators.
|
|
Clears the console before displaying to keep the table in the same place.
|
|
Args:
|
|
prices (list): A list of asset data dictionaries from get_asset_prices.
|
|
previous_prices (dict): A dictionary of previous prices with asset names as keys.
|
|
"""
|
|
clear_console()
|
|
if not prices:
|
|
print("No price data to display.")
|
|
return
|
|
|
|
# Filter prices to only include assets in assets_to_track
|
|
tracked_assets = {asset['name'] for asset in assets_to_track}
|
|
prices = [asset for asset in prices if asset['name'] in tracked_assets]
|
|
|
|
# ANSI color codes
|
|
GREEN = '\033[92m'
|
|
RED = '\033[91m'
|
|
RESET = '\033[0m'
|
|
|
|
print(f"{'Asset':<12} | {'Mark Price':<20} | {'Change'}")
|
|
print("-" * 40)
|
|
for asset in prices:
|
|
current_price = float(asset['mark_price']) if asset['mark_price'] else 0
|
|
previous_price = previous_prices.get(asset['name'], 0)
|
|
|
|
indicator = " "
|
|
color = RESET
|
|
if previous_price and current_price > previous_price:
|
|
indicator = "↑"
|
|
color = GREEN
|
|
elif previous_price and current_price < previous_price:
|
|
indicator = "↓"
|
|
color = RED
|
|
|
|
# Use precision set in assets_to_track
|
|
precision = next((a['precision'] for a in assets_to_track if a['name'] == asset['name']), 2)
|
|
price_str = f"${current_price:,.{precision}f}" if current_price else "N/A"
|
|
print(f"{asset['name']:<12} | {color}{price_str:<20}{RESET} | {color}{indicator}{RESET}")
|
|
"""
|
|
Displays a list of asset prices in a formatted table with price change indicators.
|
|
Clears the console before displaying to keep the table in the same place.
|
|
Args:
|
|
prices (list): A list of asset data dictionaries from get_asset_prices.
|
|
previous_prices (dict): A dictionary of previous prices with asset names as keys.
|
|
"""
|
|
clear_console()
|
|
if not prices:
|
|
print("No price data to display.")
|
|
return
|
|
|
|
# ANSI color codes
|
|
GREEN = '\033[92m'
|
|
RED = '\033[91m'
|
|
RESET = '\033[0m'
|
|
|
|
print("\n")
|
|
print("-" * 38)
|
|
print(f"{'Asset':<8} | {'Mark Price':<15} | {'Change':<6} |")
|
|
print("-" * 38)
|
|
for asset in prices:
|
|
current_price = float(asset['mark_price']) if asset['mark_price'] else 0
|
|
previous_price = previous_prices.get(asset['name'], 0)
|
|
|
|
indicator = " "
|
|
color = RESET
|
|
if previous_price and current_price > previous_price:
|
|
indicator = "↑"
|
|
color = GREEN
|
|
elif previous_price and current_price < previous_price:
|
|
indicator = "↓"
|
|
color = RED
|
|
|
|
# Use precision set in assets_to_track
|
|
precision = next((a['precision'] for a in assets_to_track if a['name'] == asset['name']), 2)
|
|
price_str = f"${current_price:,.{precision}f}" if current_price else "N/A"
|
|
print(f"{asset['name']:<8} | {color}{price_str:<15}{RESET} | {color}{indicator:<4}{RESET} | ")
|
|
print("-" * 38)
|
|
if __name__ == "__main__":
|
|
assets_to_track = [
|
|
{"name": "BTC", "precision": 0}
|
|
]
|
|
previous_prices = {}
|
|
|
|
while True:
|
|
# Pass only the asset names to get_asset_prices
|
|
asset_names = [a["name"] for a in assets_to_track]
|
|
current_prices_data = get_asset_prices(asset_names)
|
|
display_prices_table(current_prices_data, previous_prices)
|
|
|
|
# Update previous_prices for the next iteration
|
|
for asset in current_prices_data:
|
|
if asset['mark_price']:
|
|
previous_prices[asset['name']] = float(asset['mark_price'])
|
|
|
|
time.sleep(1) # Add a delay to avoid overwhelming the API
|
|
|