Python Script: Precious Metals & Crypto Price Fetcher

APIs Used

1. CoinGecko API (https://www.coingecko.com/api) - Free, no signup required

   • Provides cryptocurrency prices with generous rate limits (~10-50 calls/min)
   • No API key needed

2. Metals API (https://metals-api.com) - Free tier available

   • Provides precious metal prices
   • Signup required (free tier: 100 requests/month, or use alternative below)
   • Alternative: Use Yahoo Finance via yfinance library (no signup)

I'll provide two versions: one using Metals API (more accurate) and one using yfinance (no signup needed).

---

Installation

pip install requests mysql-connector-python python-dotenv schedule

For the no-signup version, add:

pip install yfinance

---

Database Schema

CREATE TABLE IF NOT EXISTS price_history (
    id INT AUTO_INCREMENT PRIMARY KEY,
    asset_name VARCHAR(50) NOT NULL,
    asset_symbol VARCHAR(20) NOT NULL,
    asset_type ENUM('crypto', 'precious_metal') NOT NULL,
    price_usd DECIMAL(20, 8) NOT NULL,
    fetched_at TIMESTAMP NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_asset_time (asset_symbol, fetched_at)
);

---

Python Script (Using YFinance - No Signup Required)

This version uses no external APIs requiring signup:

#!/usr/bin/env python3
"""
Precious Metals & Crypto Price Fetcher
Fetches prices every 60 seconds and stores in MySQL database.
Uses CoinGecko (crypto) and yfinance (metals) - no API keys required.
"""

import logging
import json
import time
import sys
from datetime import datetime
from typing import Dict, List, Tuple
import mysql.connector
from mysql.connector import Error as MySQLError
import requests
from requests.exceptions import RequestException, Timeout
import yfinance as yf

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('price_fetcher.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# Configuration
DB_CONFIG = {
    'host': 'localhost',
    'user': 'root',
    'password': 'your_password',  # Change this
    'database': 'price_tracker',
    'autocommit': False
}

CRYPTO_IDS = {
    'bitcoin': 'BTC',
    'ethereum': 'ETH',
    'binancecoin': 'BNB',
    'solana': 'SOL',
    'ripple': 'XRP'
}

METALS_SYMBOLS = {
    'GC=F': 'Gold',      # Gold futures
    'SI=F': 'Silver',    # Silver futures
    'PL=F': 'Platinum',  # Platinum futures
    'PA=F': 'Palladium'  # Palladium futures
}

API_TIMEOUT = 10  # seconds


class PriceFetcher:
    """Fetches prices from multiple sources and stores in database."""
    
    def __init__(self):
        self.db_connection = None
        self.connect_db()
    
    def connect_db(self) -> None:
        """Establish MySQL database connection."""
        try:
            self.db_connection = mysql.connector.connect(**DB_CONFIG)
            logger.info("Connected to MySQL database successfully")
        except MySQLError as e:
            logger.error(f"Database connection failed: {e}")
            raise
    
    def close_db(self) -> None:
        """Close database connection."""
        if self.db_connection and self.db_connection.is_connected():
            self.db_connection.close()
            logger.info("Database connection closed")
    
    def fetch_crypto_prices(self) -> Dict[str, float]:
        """
        Fetch cryptocurrency prices from CoinGecko API.
        
        Returns:
            Dict mapping crypto symbol to USD price
        """
        prices = {}
        try:
            # Fetch all crypto prices in one request
            crypto_names = ','.join(CRYPTO_IDS.keys())
            url = (
                f"https://api.coingecko.com/api/v3/simple/price"
                f"?ids={crypto_names}&vs_currencies=usd"
            )
            
            response = requests.get(url, timeout=API_TIMEOUT)
            response.raise_for_status()
            data = response.json()
            
            for crypto_name, symbol in CRYPTO_IDS.items():
                if crypto_name in data and 'usd' in data[crypto_name]:
                    prices[symbol] = data[crypto_name]['usd']
                    logger.debug(f"Fetched {symbol}: ${prices[symbol]:,.2f}")
            
            logger.info(f"Successfully fetched {len(prices)} crypto prices")
            return prices
        
        except Timeout:
            logger.error("Crypto API request timed out")
            return {}
        except RequestException as e:
            logger.error(f"Crypto API request failed: {e}")
            return {}
        except (json.JSONDecodeError, KeyError) as e:
            logger.error(f"Error parsing crypto response: {e}")
            return {}
    
    def fetch_metal_prices(self) -> Dict[str, float]:
        """
        Fetch precious metal prices using yfinance.
        Prices are in USD per troy ounce (for spot prices).
        
        Returns:
            Dict mapping metal symbol to USD price
        """
        prices = {}
        try:
            for symbol, metal_name in METALS_SYMBOLS.items():
                try:
                    ticker = yf.Ticker(symbol)
                    data = ticker.history(period='1d')
                    
                    if not data.empty and 'Close' in data.columns:
                        price = float(data['Close'].iloc[-1])
                        prices[metal_name] = price
                        logger.debug(f"Fetched {metal_name}: ${price:,.2f}")
                    else:
                        logger.warning(f"No data available for {metal_name}")
                
                except Exception as e:
                    logger.warning(f"Failed to fetch {metal_name}: {e}")
            
            logger.info(f"Successfully fetched {len(prices)} metal prices")
            return prices
        
        except Exception as e:
            logger.error(f"Metal price fetching failed: {e}")
            return {}
    
    def store_prices(self, prices: Dict[str, Tuple[float, str]]) -> None:
        """
        Store fetched prices in database.
        
        Args:
            prices: Dict of {symbol: (price, asset_type)}
        """
        if not prices:
            logger.warning("No prices to store")
            return
        
        try:
            cursor = self.db_connection.cursor()
            insert_query = """
                INSERT INTO price_history 
                (asset_name, asset_symbol, asset_type, price_usd, fetched_at)
                VALUES (%s, %s, %s, %s, %s)
            """
            
            timestamp = datetime.utcnow()
            records = []
            
            for symbol, (price, asset_type) in prices.items():
                # Determine full asset name
                asset_name = (
                    CRYPTO_IDS.get(symbol.lower()) or 
                    next((v for k, v in METALS_SYMBOLS.items() if v == symbol), symbol)
                )
                
                records.append((asset_name, symbol, asset_type, price, timestamp))
            
            cursor.executemany(insert_query, records)
            self.db_connection.commit()
            logger.info(f"Stored {len(records)} price records in database")
            cursor.close()
        
        except MySQLError as e:
            logger.error(f"Database insert failed: {e}")
            try:
                self.db_connection.rollback()
            except MySQLError:
                logger.error("Rollback failed, attempting to reconnect")
                self.close_db()
                self.connect_db()
    
    def fetch_and_store(self) -> None:
        """Fetch all prices and store in database."""
        try:
            all_prices = {}
            
            # Fetch cryptocurrencies
            crypto_prices = self.fetch_crypto_prices()
            for symbol, price in crypto_prices.items():
                all_prices[symbol] = (price, 'crypto')
            
            # Fetch precious metals
            metal_prices = self.fetch_metal_prices()
            for symbol, price in metal_prices.items():
                all_prices[symbol] = (price, 'precious_metal')
            
            # Store in database
            self.store_prices(all_prices)
        
        except Exception as e:
            logger.error(f"Unexpected error in fetch_and_store: {e}", exc_info=True)
    
    def run(self, interval: int = 60) -> None:
        """
        Run the price fetcher indefinitely.
        
        Args:
            interval: Time in seconds between fetches (default: 60)
        """
        logger.info(f"Starting price fetcher (interval: {interval}s)")
        
        try:
            while True:
                logger.info("=" * 50)
                self.fetch_and_store()
                logger.info(f"Next fetch in {interval} seconds")
                time.sleep(interval)
        
        except KeyboardInterrupt:
            logger.info("Received interrupt signal, shutting down gracefully")
        except Exception as e:
            logger.error(f"Fatal error: {e}", exc_info=True)
        finally:
            self.close_db()
            logger.info("Price fetcher stopped")


def main():
    """Main entry point."""
    fetcher = PriceFetcher()
    fetcher.run(interval=60)


if __name__ == '__main__':
    main()

---

Configuration Steps

1. Create MySQL Database

CREATE DATABASE IF NOT EXISTS price_tracker;
USE price_tracker;

CREATE TABLE IF NOT EXISTS price_history (
    id INT AUTO_INCREMENT PRIMARY KEY,
    asset_name VARCHAR(50) NOT NULL,
    asset_symbol VARCHAR(20) NOT NULL,
    asset_type ENUM('crypto', 'precious_metal') NOT NULL,
    price_usd DECIMAL(20, 8) NOT NULL,
    fetched_at TIMESTAMP NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_asset_time (asset_symbol, fetched_at)
);

2. Update Database Credentials

Edit the DB_CONFIG in the script:

DB_CONFIG = {
    'host': 'localhost',
    'user': 'your_mysql_user',
    'password': 'your_mysql_password',
    'database': 'price_tracker'
}

3. Run the Script

python price_fetcher.py

---

Helpful Queries

-- Last price for each asset
SELECT 
    asset_symbol, 
    price_usd, 
    fetched_at
FROM price_history
WHERE (asset_symbol, fetched_at) IN (
    SELECT asset_symbol, MAX(fetched_at)
    FROM price_history
    GROUP BY asset_symbol
)
ORDER BY asset_symbol;

-- Price history for Bitcoin in last hour
SELECT * FROM price_history
WHERE asset_symbol = 'BTC'
  AND fetched_at > DATE_SUB(NOW(), INTERVAL 1 HOUR)
ORDER BY fetched_at DESC;

-- Average daily price
SELECT 
    asset_symbol,
    DATE(fetched_at) as date,
    AVG(price_usd) as avg_price,
    MIN(price_usd) as min_price,
    MAX(price_usd) as max_price
FROM price_history
GROUP BY asset_symbol, DATE(fetched_at)
ORDER BY date DESC;

---

Notes

• yfinance provides futures prices (GC=F, SI=F, etc.) which are correlated with spot prices but not identical
• CoinGecko is free with very generous rate limits—perfect for crypto
• Logs are written to price_fetcher.log and console
• The script handles disconnections gracefully and attempts to reconnect
• For production, consider using a process manager like supervisord or systemd