131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
import argparse
|
|
from datetime import datetime
|
|
|
|
from hyperliquid.info import Info
|
|
from hyperliquid.utils import constants
|
|
from hyperliquid.utils.error import ClientError
|
|
|
|
# Assuming logging_utils.py is in the same directory
|
|
from logging_utils import setup_logging
|
|
|
|
class MarketDataFeeder:
|
|
"""
|
|
Fetches live market prices for all coins from Hyperliquid, displays them
|
|
in a table, and saves the data for other scripts to use.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.info = Info(constants.MAINNET_API_URL, skip_ws=True)
|
|
self.coins = self._load_coins()
|
|
self.data_folder = "_data"
|
|
self.price_file = os.path.join(self.data_folder, "current_prices.json")
|
|
|
|
def _load_coins(self) -> list:
|
|
"""Loads the list of coins from the coin_precision.json file."""
|
|
try:
|
|
with open("_data/coin_precision.json", 'r') as f:
|
|
coins_data = json.load(f)
|
|
logging.info(f"Loaded {len(coins_data)} coins from 'coin_precision.json'.")
|
|
return list(coins_data.keys())
|
|
except FileNotFoundError:
|
|
logging.error("'coin_precision.json' not found. Please run list_coins.py first.")
|
|
sys.exit(1)
|
|
except (IOError, json.JSONDecodeError) as e:
|
|
logging.error(f"Failed to load or parse 'coin_precision.json': {e}")
|
|
sys.exit(1)
|
|
|
|
def fetch_and_update_prices(self):
|
|
"""Fetches all asset contexts and updates the local price dictionary."""
|
|
try:
|
|
meta, asset_contexts = self.info.meta_and_asset_ctxs()
|
|
|
|
prices = {}
|
|
# The API returns two lists. The first has static info (name), the second has live data (markPx).
|
|
# We need to pair them up. They are guaranteed to be in the same order.
|
|
for asset_info, asset_ctx in zip(meta["universe"], asset_contexts):
|
|
coin_name = asset_info["name"]
|
|
if asset_ctx: # The context can sometimes be null for unlisted assets
|
|
mark_price = asset_ctx["markPx"]
|
|
prices[coin_name] = float(mark_price)
|
|
|
|
return prices
|
|
|
|
except ClientError as e:
|
|
logging.error(f"An API error occurred: {e}")
|
|
except Exception as e:
|
|
logging.error(f"An unexpected error occurred during fetch: {e}")
|
|
return None
|
|
|
|
def save_prices_to_file(self, prices: dict):
|
|
"""Saves the current price dictionary to a JSON file."""
|
|
if not os.path.exists(self.data_folder):
|
|
os.makedirs(self.data_folder)
|
|
|
|
try:
|
|
with open(self.price_file, 'w', encoding='utf-8') as f:
|
|
json.dump(prices, f, indent=4)
|
|
logging.debug(f"Prices successfully saved to '{self.price_file}'")
|
|
except IOError as e:
|
|
logging.error(f"Failed to write to price file: {e}")
|
|
|
|
def display_prices(self, prices: dict):
|
|
"""Clears the screen and displays a formatted table of all prices."""
|
|
# Use ANSI escape codes to clear the screen and move the cursor to the top-left
|
|
print("\x1b[H\x1b[J", end="")
|
|
|
|
print("--- Hyperliquid Market Prices ---")
|
|
print(f"Last Updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
table_width = 40
|
|
print("-" * table_width)
|
|
print(f"{'#':<4} | {'Coin':<10} | {'Price':>20}")
|
|
print("-" * table_width)
|
|
|
|
# Sort prices by coin name for a consistent display
|
|
sorted_prices = sorted(prices.items())
|
|
|
|
for i, (coin, price) in enumerate(sorted_prices, 1):
|
|
print(f"{i:<4} | {coin:<10} | {price:>20}")
|
|
|
|
print("-" * table_width)
|
|
sys.stdout.flush()
|
|
|
|
def run(self):
|
|
"""Main loop to fetch, display, and save prices every second."""
|
|
logging.info("Starting market data feed. Press Ctrl+C to stop.")
|
|
while True:
|
|
current_prices = self.fetch_and_update_prices()
|
|
if current_prices:
|
|
self.save_prices_to_file(current_prices)
|
|
|
|
# Only display the table if logging is set to DEBUG
|
|
if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
|
|
self.display_prices(current_prices)
|
|
|
|
time.sleep(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Fetch live price data from Hyperliquid.")
|
|
parser.add_argument(
|
|
"--log-level",
|
|
default="normal",
|
|
choices=['off', 'normal', 'debug'],
|
|
help="Set the logging level for the script."
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
setup_logging(args.log_level, 'MarketFeed')
|
|
|
|
feeder = MarketDataFeeder()
|
|
try:
|
|
feeder.run()
|
|
except KeyboardInterrupt:
|
|
logging.info("Market data feeder stopped.")
|
|
sys.exit(0)
|
|
|