""" Fetch ASIC miner prices from Hashrate Index API Downloads price indices for efficiency categories and calculates individual miner prices """ import requests import pandas as pd import os from datetime import datetime from miner_specs import MINER_SPECS # Efficiency category mapping for each miner EFFICIENCY_MAPPING = { 's19pro': '25to38', # 30 W/TH 's19jpro': '25to38', # 30 W/TH 's19kpro': '19to25', # 23 W/TH 's21': 'under19', # 18 W/TH 'ka3': 'under19', # 19 W/TH 't19': '25to38', # 38 W/TH 's19xp': '19to25', # 21 W/TH 's19apro': '25to38', # 31 W/TH 'm50s': '19to25', # 24 W/TH 'm53': '25to38' , # 29 W/TH 'm30s':'25to38' # 31 W/TH } def fetch_asic_price_for_date(target_date, api_key=None, currency='USD'): """ Fetch ASIC price for a specific historical date Parameters: ----------- target_date : str or datetime Target date in format 'YYYY-MM-DD' or datetime object api_key : str, optional Hashrate Index API key currency : str Currency (default: USD) Returns: -------- tuple: (dict of miner_name -> price, bool indicating if data exists) """ # Get API key from environment if not provided if not api_key: api_key = os.environ.get('HASHRATE_API_KEY') if not api_key: print("⚠️ No API key found, using fallback prices") return get_fallback_prices(), False # Convert to datetime if string if isinstance(target_date, str): target_date = pd.to_datetime(target_date) url = "https://api.hashrateindex.com/v1/hashrateindex/asic/price-index" headers = { "Accept": "application/json", "X-Hi-Api-Key": api_key } params = { "currency": currency, "span": "ALL" # Get all historical data } try: response = requests.get(url, headers=headers, params=params, timeout=30) response.raise_for_status() data = response.json().get("data", []) if not data: print("⚠️ No data returned from API") return get_fallback_prices(), False # Create DataFrame df = pd.DataFrame(data) df['timestamp'] = pd.to_datetime(df['timestamp']) df['date'] = df['timestamp'].dt.date # Find closest date target_date_only = target_date.date() df['date_diff'] = abs((df['timestamp'].dt.date - target_date_only).apply(lambda x: x.days)) # Get row with closest date (within 7 days tolerance) closest_row = df.loc[df['date_diff'].idxmin()] if closest_row['date_diff'] > 7: print(f"⚠️ No price data within 7 days of {target_date_only}") return get_fallback_prices(), False # Extract efficiency prices efficiency_prices = { 'under19': closest_row.get('under19', None), '19to25': closest_row.get('19to25', None), '25to38': closest_row.get('25to38', None) } print(f"✅ Found price data for {closest_row['date']} (requested: {target_date_only})") print(f" under19: ${efficiency_prices['under19']}/TH") print(f" 19to25: ${efficiency_prices['19to25']}/TH") print(f" 25to38: ${efficiency_prices['25to38']}/TH") # Calculate miner prices miner_prices = {} for miner_name, efficiency_cat in EFFICIENCY_MAPPING.items(): if miner_name in MINER_SPECS: hashrate = MINER_SPECS[miner_name]['hashrate'] price_per_th = efficiency_prices.get(efficiency_cat) if price_per_th and price_per_th > 0: miner_prices[miner_name] = hashrate * price_per_th else: miner_prices[miner_name] = get_fallback_price_for_miner(miner_name) return miner_prices, True except requests.exceptions.RequestException as e: print(f"⚠️ API request failed: {e}") return get_fallback_prices(), False except Exception as e: print(f"⚠️ Error: {e}") return get_fallback_prices(), False def fetch_asic_price_index(api_key=None, currency='USD'): """ Fetch ASIC price index from Hashrate Index API Parameters: ----------- api_key : str, optional Hashrate Index API key (if None, returns fallback prices) currency : str Currency (default: USD) Returns: -------- dict Dictionary of miner_name -> price (USD) """ if not api_key: api_key = os.environ.get('HASHRATE_API_KEY') if not api_key: print("⚠️ No API key provided, using fallback prices") return get_fallback_prices() url = "https://api.hashrateindex.com/v1/hashrateindex/asic/price-index" headers = { "Accept": "application/json", "X-Hi-Api-Key": api_key } params = { "currency": currency, "span": "1Y" # Last year of data } try: response = requests.get(url, headers=headers, params=params, timeout=30) response.raise_for_status() data = response.json().get("data", []) if not data: print("⚠️ No data returned from API, using fallback prices") return get_fallback_prices() # Get most recent data df = pd.DataFrame(data) df['timestamp'] = pd.to_datetime(df['timestamp']) df = df.sort_values('timestamp', ascending=False) # Get latest row latest = df.iloc[0] # Extract efficiency category prices ($/TH) efficiency_prices = { 'under19': latest.get('under19', None), '19to25': latest.get('19to25', None), '25to38': latest.get('25to38', None) } print(f"✅ Fetched price index (date: {latest['timestamp'].date()})") print(f" under19: ${efficiency_prices['under19']}/TH") print(f" 19to25: ${efficiency_prices['19to25']}/TH") print(f" 25to38: ${efficiency_prices['25to38']}/TH") # Calculate individual miner prices miner_prices = {} for miner_name, efficiency_cat in EFFICIENCY_MAPPING.items(): if miner_name in MINER_SPECS: hashrate = MINER_SPECS[miner_name]['hashrate'] price_per_th = efficiency_prices.get(efficiency_cat) if price_per_th: miner_prices[miner_name] = hashrate * price_per_th else: # Fallback if category not available miner_prices[miner_name] = get_fallback_price_for_miner(miner_name) return miner_prices except requests.exceptions.RequestException as e: print(f"⚠️ API request failed: {e}") print(" Using fallback prices") return get_fallback_prices() except Exception as e: print(f"⚠️ Error processing API data: {e}") print(" Using fallback prices") return get_fallback_prices() def get_fallback_prices(): """ Fallback prices when API is unavailable Based on approximate market prices as of Dec 2024 Returns: -------- dict Dictionary of miner_name -> price (USD) """ return { 's19pro': 2500, 's19jpro': 2200, 's19kpro': 3500, 's21': 5500, 'ka3': 5000, 't19': 2000, 's19xp': 4000, 's19apro': 2800, 'm50s': 3200, 'm53': 8000 } def get_fallback_price_for_miner(miner_name): """Get fallback price for a specific miner""" fallback = get_fallback_prices() return fallback.get(miner_name, 2500) # For backward compatibility with your existing code FALLBACK_PRICES = get_fallback_prices() if __name__ == "__main__": print("\n" + "="*60) print("ASIC Price Fetcher Test") print("="*60 + "\n") # Try to get API key from environment variable api_key = os.environ.get('HASHRATE_API_KEY') if api_key: print(f"Using API key: {api_key[:8]}...") else: print("No API key found in HASHRATE_API_KEY environment variable") print("Set it with: export HASHRATE_API_KEY='your-key-here'") # Fetch prices prices = fetch_asic_price_index(api_key) print("\n" + "="*60) print("Current Miner Prices") print("="*60) for miner_name, price in sorted(prices.items()): specs = MINER_SPECS.get(miner_name, {}) full_name = specs.get('full_name', miner_name) hashrate = specs.get('hashrate', 0) efficiency = specs.get('efficiency', 0) efficiency_cat = EFFICIENCY_MAPPING.get(miner_name, 'unknown') print(f"{full_name:25s} ({hashrate:3.0f} TH/s, {efficiency:4.1f} W/TH)") print(f" Category: {efficiency_cat:15s} → Price: ${price:,.2f}") print() print("="*60)