ditus/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

save market cap of all coins

Browse Source
This commit is contained in:
DiTus
2025-10-21 23:52:32 +02:00
parent 5a05f0d190
commit 75c0cc77cc
2 changed files with 116 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
coin_id_map.py Normal file
View File

@ -0,0 +1,80 @@
import os
import json
import logging
import requests
from hyperliquid.info import Info
from hyperliquid.utils import constants
from logging_utils import setup_logging
def update_coin_mapping():
"""
Fetches all assets from Hyperliquid and all coins from CoinGecko,
then creates and saves a mapping from the Hyperliquid symbol to the
CoinGecko ID.
"""
setup_logging('normal', 'CoinMapUpdater')
logging.info("Starting coin mapping update process...")
# --- 1. Fetch all assets from Hyperliquid ---
try:
logging.info("Fetching assets from Hyperliquid...")
info = Info(constants.MAINNET_API_URL, skip_ws=True)
# The meta object contains the 'universe' list with asset details
meta, asset_contexts = info.meta_and_asset_ctxs()
# --- FIX: The asset names are in the 'universe' list inside the meta object ---
# The 'universe' is a list of dictionaries, each with a 'name'
hyperliquid_assets = [asset['name'] for asset in meta['universe']]
logging.info(f"Found {len(hyperliquid_assets)} assets on Hyperliquid.")
except Exception as e:
logging.error(f"Failed to fetch assets from Hyperliquid: {e}")
return
# --- 2. Fetch all coins from CoinGecko ---
try:
logging.info("Fetching coin list from CoinGecko...")
response = requests.get("https://api.coingecko.com/api/v3/coins/list")
response.raise_for_status()
coingecko_coins = response.json()
# Create a lookup table: {symbol: id}
coingecko_lookup = {coin['symbol'].upper(): coin['id'] for coin in coingecko_coins}
logging.info(f"Found {len(coingecko_coins)} coins on CoinGecko.")
except requests.exceptions.RequestException as e:
logging.error(f"Failed to fetch coin list from CoinGecko: {e}")
return
# --- 3. Create the mapping ---
final_mapping = {}
manual_overrides = {
"HYPE": "hyperliquid",
"PUMP": "pump-fun",
"ASTER": "astar",
}
logging.info("Generating symbol-to-id mapping...")
for asset_symbol in hyperliquid_assets:
# Check for manual overrides first
if asset_symbol in manual_overrides:
final_mapping[asset_symbol] = manual_overrides[asset_symbol]
continue
# Try to find a direct match in the CoinGecko lookup table
if asset_symbol in coingecko_lookup:
final_mapping[asset_symbol] = coingecko_lookup[asset_symbol]
else:
logging.warning(f"No direct match found for '{asset_symbol}' on CoinGecko. It will be excluded.")
# --- 4. Save the mapping to a file ---
map_file_path = os.path.join("_data", "coin_id_map.json")
try:
with open(map_file_path, 'w', encoding='utf-8') as f:
json.dump(final_mapping, f, indent=4, sort_keys=True)
logging.info(f"Successfully saved new coin mapping with {len(final_mapping)} entries to '{map_file_path}'.")
except IOError as e:
logging.error(f"Failed to write coin mapping file: {e}")
if __name__ == "__main__":
update_coin_mapping()

73
market_cap_fetcher.py
View File

@ -8,47 +8,52 @@ import requests
import time
from datetime import datetime, timezone, timedelta
import json
from dotenv import load_dotenv
load_dotenv()
# Assuming logging_utils.py is in the same directory
from logging_utils import setup_logging
class MarketCapFetcher:
"""
Fetches historical daily market cap data from the CoinGecko API and
intelligently updates the SQLite database. It processes individual coins,
aggregates stablecoins, and captures total market cap metrics.
intelligently updates the SQLite database for all coins found in the coin map.
"""
COIN_ID_MAP = {
"BTC": "bitcoin",
"ETH": "ethereum",
"SOL": "solana",
"BNB": "binancecoin",
"HYPE": "hyperliquid",
"ASTER": "astar",
"ZEC": "zcash",
"PUMP": "pump-fun", # Correct ID is 'pump-fun'
"SUI": "sui"
}
STABLECOIN_ID_MAP = {
"USDT": "tether",
"USDC": "usd-coin",
"USDE": "ethena-usde",
"DAI": "dai",
"PYUSD": "paypal-usd"
}
def __init__(self, log_level: str, coins: list):
def __init__(self, log_level: str):
setup_logging(log_level, 'MarketCapFetcher')
self.coins_to_fetch = coins
self.db_path = os.path.join("_data", "market_data.db")
self.api_base_url = "https://api.coingecko.com/api/v3"
self.api_key = os.environ.get("COINGECKO_API_KEY")
if not self.api_key:
logging.error("CoinGecko API key not found. Please set the COINGECKO_API_KEY environment variable.")
sys.exit(1)
self.COIN_ID_MAP = self._load_coin_id_map()
if not self.COIN_ID_MAP:
logging.error("Coin ID map is empty. Run 'update_coin_map.py' to generate it.")
sys.exit(1)
# --- FIX: The list of coins to fetch is now all coins from the map ---
self.coins_to_fetch = list(self.COIN_ID_MAP.keys())
self.STABLECOIN_ID_MAP = {
"USDT": "tether",
"USDC": "usd-coin",
"USDE": "ethena-usde",
"DAI": "dai",
"PYUSD": "paypal-usd"
}
def _load_coin_id_map(self) -> dict:
"""Loads the dynamically generated coin-to-id mapping."""
map_file_path = os.path.join("_data", "coin_id_map.json")
try:
with open(map_file_path, 'r') as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.error(f"Could not load '{map_file_path}'. Please run 'update_coin_map.py' first. Error: {e}")
return {}
def run(self):
"""
@ -58,7 +63,7 @@ class MarketCapFetcher:
with sqlite3.connect(self.db_path) as conn:
conn.execute("PRAGMA journal_mode=WAL;")
# 1. Process individual coins
# 1. Process individual coins from the map
for coin_symbol in self.coins_to_fetch:
coin_id = self.COIN_ID_MAP.get(coin_symbol.upper())
if not coin_id:
@ -123,7 +128,6 @@ class MarketCapFetcher:
table_name = "TOTAL_market_cap_daily"
try:
# --- FIX: Use the current date instead of yesterday's ---
today_date = datetime.now(timezone.utc).date()
cursor = conn.cursor()
@ -131,7 +135,6 @@ class MarketCapFetcher:
table_exists = cursor.fetchone()
if table_exists:
# Check if we already have a record for today
cursor.execute(f"SELECT 1 FROM \"{table_name}\" WHERE date(datetime_utc) = ? LIMIT 1", (today_date.isoformat(),))
if cursor.fetchone():
logging.info(f"Total market cap for {today_date} already exists. Skipping.")
@ -245,7 +248,7 @@ class MarketCapFetcher:
try:
logging.debug(f"Fetching last {days} days for {coin_id}...")
response = requests.get(url, headers=headers)
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
@ -264,12 +267,7 @@ class MarketCapFetcher:
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Fetch historical market cap data from CoinGecko.")
parser.add_argument(
"--coins",
nargs='+',
default=["BTC", "ETH", "SOL", "BNB", "HYPE", "ASTER", "ZEC", "PUMP", "SUI"],
help="List of coin symbols to fetch (e.g., BTC ETH)."
)
# --- FIX: The --coins argument is no longer needed as the script is now fully automated ---
parser.add_argument(
"--log-level",
default="normal",
@ -278,6 +276,7 @@ if __name__ == "__main__":
)
args = parser.parse_args()
fetcher = MarketCapFetcher(log_level=args.log_level, coins=args.coins)
# The 'coins' argument is no longer passed to the constructor
fetcher = MarketCapFetcher(log_level=args.log_level)
fetcher.run()