Python Price Tracker for Precious Metals & Cryptocurrencies

Overview

This script fetches precious metal and cryptocurrency prices every 60 seconds and stores them in MySQL. It uses two free APIs:

• Metals API: metals-api.com - Free tier (100 requests/month). Note: This is quite limited, so I'll also provide an alternative using goldapi.io which offers 100 requests/month for free.
• CoinGecko API: coingecko.com/api - No signup required, generous free tier (10-50 calls/minute)

Best Alternative (recommended): Use CoinGecko for crypto AND metals since they support precious metals too!

Required Python Packages

pip install requests mysql-connector-python python-dotenv

MySQL Database Schema

-- Create the database
CREATE DATABASE IF NOT EXISTS price_tracker;
USE price_tracker;

-- Create the prices table
CREATE TABLE IF NOT EXISTS asset_prices (
    id INT AUTO_INCREMENT PRIMARY KEY,
    asset_symbol VARCHAR(10) NOT NULL,
    asset_name VARCHAR(50) NOT NULL,
    asset_type ENUM('metal', 'crypto') NOT NULL,
    price_usd DECIMAL(20, 8) NOT NULL,
    timestamp DATETIME NOT NULL,
    INDEX idx_symbol_timestamp (asset_symbol, timestamp),
    INDEX idx_timestamp (timestamp)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

Python Script

import requests
import mysql.connector
from mysql.connector import Error
import time
import logging
from datetime import datetime
from typing import Dict, List, Optional
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('price_tracker.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# Database configuration
DB_CONFIG = {
    'host': os.getenv('DB_HOST', 'localhost'),
    'user': os.getenv('DB_USER', 'root'),
    'password': os.getenv('DB_PASSWORD', 'your_password'),
    'database': os.getenv('DB_NAME', 'price_tracker')
}

# Asset configurations
CRYPTO_ASSETS = {
    'bitcoin': 'BTC',
    'ethereum': 'ETH',
    'binancecoin': 'BNB',
    'solana': 'SOL',
    'ripple': 'XRP'
}

METAL_ASSETS = {
    'gold': 'XAU',
    'silver': 'XAG',
    'platinum': 'XPT',
    'palladium': 'XPD'
}

# API endpoints
COINGECKO_BASE_URL = 'https://api.coingecko.com/api/v3'


class PriceTracker:
    """Main class for tracking and storing asset prices"""
    
    def __init__(self):
        self.connection = None
        self.connect_to_database()
    
    def connect_to_database(self) -> None:
        """Establish connection to MySQL database"""
        try:
            self.connection = mysql.connector.connect(**DB_CONFIG)
            if self.connection.is_connected():
                logger.info("Successfully connected to MySQL database")
        except Error as e:
            logger.error(f"Error connecting to MySQL: {e}")
            raise
    
    def ensure_connection(self) -> None:
        """Ensure database connection is alive, reconnect if needed"""
        try:
            if self.connection is None or not self.connection.is_connected():
                logger.warning("Database connection lost, reconnecting...")
                self.connect_to_database()
        except Error as e:
            logger.error(f"Error ensuring connection: {e}")
    
    def fetch_crypto_prices(self) -> List[Dict]:
        """Fetch cryptocurrency prices from CoinGecko API"""
        try:
            # CoinGecko API endpoint for multiple coins
            ids = ','.join(CRYPTO_ASSETS.keys())
            url = f'{COINGECKO_BASE_URL}/simple/price'
            params = {
                'ids': ids,
                'vs_currencies': 'usd'
            }
            
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            prices = []
            timestamp = datetime.now()
            
            for coin_id, symbol in CRYPTO_ASSETS.items():
                if coin_id in data and 'usd' in data[coin_id]:
                    prices.append({
                        'symbol': symbol,
                        'name': coin_id.capitalize(),
                        'type': 'crypto',
                        'price': data[coin_id]['usd'],
                        'timestamp': timestamp
                    })
            
            logger.info(f"Successfully fetched {len(prices)} crypto prices")
            return prices
            
        except requests.exceptions.RequestException as e:
            logger.error(f"Error fetching crypto prices: {e}")
            return []
        except Exception as e:
            logger.error(f"Unexpected error fetching crypto prices: {e}")
            return []
    
    def fetch_metal_prices(self) -> List[Dict]:
        """Fetch precious metal prices from CoinGecko API (alternative method)"""
        try:
            # CoinGecko also tracks precious metals as crypto tokens
            # We'll use a conversion factor approach or goldapi.io
            # For simplicity, using metalpriceapi.com free tier (no auth needed for basic)
            
            # Alternative: Using goldapi.io (requires free API key)
            # For this example, I'll use a simpler approach with metals-api.com format
            
            # Since free metal APIs are limited, here's a practical approach:
            # Use CoinGecko's precious metal tracking tokens like PAX Gold (PAXG)
            # Or use a free API with limited calls
            
            # Free option: metals.live (no auth needed but limited)
            url = 'https://www.metals.live/api/v1/spot'
            
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            prices = []
            timestamp = datetime.now()
            
            # Parse metals.live response format
            metal_mapping = {
                'gold': ('XAU', 'Gold'),
                'silver': ('XAG', 'Silver'),
                'platinum': ('XPT', 'Platinum'),
                'palladium': ('XPD', 'Palladium')
            }
            
            for metal_key, (symbol, name) in metal_mapping.items():
                if metal_key in data and 'price' in data[metal_key]:
                    prices.append({
                        'symbol': symbol,
                        'name': name,
                        'type': 'metal',
                        'price': data[metal_key]['price'],
                        'timestamp': timestamp
                    })
            
            logger.info(f"Successfully fetched {len(prices)} metal prices")
            return prices
            
        except requests.exceptions.RequestException as e:
            logger.error(f"Error fetching metal prices: {e}")
            return []
        except Exception as e:
            logger.error(f"Unexpected error fetching metal prices: {e}")
            return []
    
    def store_prices(self, prices: List[Dict]) -> None:
        """Store fetched prices in MySQL database"""
        if not prices:
            logger.warning("No prices to store")
            return
        
        try:
            self.ensure_connection()
            cursor = self.connection.cursor()
            
            insert_query = """
                INSERT INTO asset_prices 
                (asset_symbol, asset_name, asset_type, price_usd, timestamp)
                VALUES (%s, %s, %s, %s, %s)
            """
            
            records = [
                (
                    price['symbol'],
                    price['name'],
                    price['type'],
                    price['price'],
                    price['timestamp']
                )
                for price in prices
            ]
            
            cursor.executemany(insert_query, records)
            self.connection.commit()
            
            logger.info(f"Successfully stored {cursor.rowcount} price records")
            cursor.close()
            
        except Error as e:
            logger.error(f"Error storing prices in database: {e}")
            if self.connection:
                self.connection.rollback()
    
    def run(self) -> None:
        """Main loop to continuously fetch and store prices"""
        logger.info("Starting price tracker...")
        
        while True:
            try:
                # Fetch all prices
                crypto_prices = self.fetch_crypto_prices()
                metal_prices = self.fetch_metal_prices()
                
                # Combine all prices
                all_prices = crypto_prices + metal_prices
                
                # Store in database
                if all_prices:
                    self.store_prices(all_prices)
                    logger.info(f"Cycle complete. Stored {len(all_prices)} prices.")
                else:
                    logger.warning("No prices fetched in this cycle")
                
                # Wait 60 seconds before next fetch
                logger.info("Waiting 60 seconds until next fetch...")
                time.sleep(60)
                
            except KeyboardInterrupt:
                logger.info("Received shutdown signal. Stopping...")
                break
            except Exception as e:
                logger.error(f"Unexpected error in main loop: {e}")
                logger.info("Continuing after error... waiting 60 seconds")
                time.sleep(60)
        
        # Cleanup
        if self.connection and self.connection.is_connected():
            self.connection.close()
            logger.info("Database connection closed")


def main():
    """Entry point for the script"""
    tracker = PriceTracker()
    tracker.run()


if __name__ == "__main__":
    main()

Environment Configuration (.env file)

Create a .env file in the same directory:

DB_HOST=localhost
DB_USER=root
DB_PASSWORD=your_mysql_password
DB_NAME=price_tracker

Alternative: Using goldapi.io for Metals

If you prefer a dedicated metals API, here's the modified metal fetching function:

def fetch_metal_prices_goldapi(self) -> List[Dict]:
    """Fetch precious metal prices from goldapi.io (requires free API key)"""
    API_KEY = os.getenv('GOLDAPI_KEY', 'your_api_key_here')
    
    try:
        metals = ['XAU', 'XAG', 'XPT', 'XPD']
        prices = []
        timestamp = datetime.now()
        
        for metal in metals:
            url = f'https://www.goldapi.io/api/{metal}/USD'
            headers = {'x-access-token': API_KEY}
            
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            prices.append({
                'symbol': metal,
                'name': data.get('metal', metal),
                'type': 'metal',
                'price': data['price'],
                'timestamp': timestamp
            })
            
            time.sleep(0.5)  # Rate limiting
        
        return prices
        
    except Exception as e:
        logger.error(f"Error fetching metal prices: {e}")
        return []

Usage Instructions

1. Install MySQL and create the database using the SQL schema above
2. Install Python packages: pip install requests mysql-connector-python python-dotenv
3. Create .env file with your MySQL credentials
4. Run the script: python price_tracker.py

Notes

• API Limitations: Free tiers have limits. CoinGecko allows 10-50 calls/min. For metals, consider rotating between free APIs or using cached data.
• Error Handling: The script logs errors but continues running
• Database Growth: Consider adding a cleanup job to archive old data
• Production: Add monitoring, alerts, and consider using connection pooling

The script will run indefinitely, logging all activity to both console and price_tracker.log file.

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

Precious Metals & Crypto Price Tracker

APIs Used

Why these? CoinGecko's /simple/price endpoint is fully open. For metals, truly free no-key APIs are essentially extinct in 2024-2025. Metals.dev offers the most generous free tier. Alternative: GoldAPI.io (free tier, 300 req/month). I provide a fallback architecture so you can swap providers easily.

---

Required Packages

pip install requests mysql-connector-python python-dotenv

---

MySQL Table Schema

-- Run this in MySQL before starting the script

CREATE DATABASE IF NOT EXISTS price_tracker
  CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

USE price_tracker;

CREATE TABLE IF NOT EXISTS asset_prices (
    id          BIGINT       AUTO_INCREMENT PRIMARY KEY,
    asset_name  VARCHAR(50)  NOT NULL COMMENT 'Human-readable name (e.g. Gold, Bitcoin)',
    symbol      VARCHAR(10)  NOT NULL COMMENT 'Ticker symbol (e.g. XAU, BTC)',
    asset_type  ENUM('metal', 'crypto') NOT NULL,
    price_usd   DECIMAL(18, 8) NOT NULL COMMENT 'Price in USD',
    fetched_at  DATETIME     NOT NULL COMMENT 'Timestamp when price was fetched',
    created_at  DATETIME     DEFAULT CURRENT_TIMESTAMP,

    INDEX idx_symbol_time (symbol, fetched_at),
    INDEX idx_type_time   (asset_type, fetched_at)
) ENGINE=InnoDB;

---

.env File

Create a .env file in the same directory as the script:

MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=your_password_here
MYSQL_DATABASE=price_tracker

# Sign up free at https://metals.dev — paste your API key here
METALS_API_KEY=your_metals_dev_api_key_here

---

The Python Script

#!/usr/bin/env python3
"""
price_tracker.py
~~~~~~~~~~~~~~~~
Continuously fetches precious metal & top crypto prices every 60 seconds
and stores them in a MySQL database.

APIs:
  - Crypto:  CoinGecko (free, no key)
  - Metals:  Metals.dev (free tier, key required — sign up at https://metals.dev)

Usage:
  1. pip install requests mysql-connector-python python-dotenv
  2. Create MySQL database using the schema in the README
  3. Fill in .env with your credentials
  4. python price_tracker.py
"""

import os
import sys
import time
import logging
from datetime import datetime, timezone
from decimal import Decimal

import requests
import mysql.connector
from mysql.connector import Error as MySQLError
from dotenv import load_dotenv

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

load_dotenv()  # Load .env file

# Polling interval in seconds
POLL_INTERVAL = 60

# MySQL connection parameters
DB_CONFIG = {
    "host":     os.getenv("MYSQL_HOST", "localhost"),
    "port":     int(os.getenv("MYSQL_PORT", 3306)),
    "user":     os.getenv("MYSQL_USER", "root"),
    "password": os.getenv("MYSQL_PASSWORD", ""),
    "database": os.getenv("MYSQL_DATABASE", "price_tracker"),
}

# Metals.dev API key (free tier: 50 requests/day)
METALS_API_KEY = os.getenv("METALS_API_KEY", "")

# CoinGecko endpoint — no key needed
COINGECKO_URL = "https://api.coingecko.com/api/v3/simple/price"

# Metals.dev endpoint
METALS_URL = "https://api.metals.dev/v1/latest"

# Mapping: CoinGecko IDs → (display_name, symbol)
CRYPTO_ASSETS = {
    "bitcoin":  ("Bitcoin",  "BTC"),
    "ethereum": ("Ethereum", "ETH"),
    "binancecoin": ("BNB",   "BNB"),
    "solana":   ("Solana",   "SOL"),
    "ripple":   ("XRP",      "XRP"),
}

# Mapping: metals.dev keys → (display_name, symbol)
# metals.dev returns keys like "gold", "silver", "platinum", "palladium"
METAL_ASSETS = {
    "gold":      ("Gold",      "XAU"),
    "silver":    ("Silver",    "XAG"),
    "platinum":  ("Platinum",  "XPT"),
    "palladium": ("Palladium", "XPD"),
}

# Request timeout (seconds)
REQUEST_TIMEOUT = 15

# ---------------------------------------------------------------------------
# Logging setup
# ---------------------------------------------------------------------------

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler("price_tracker.log", encoding="utf-8"),
    ],
)
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Database helpers
# ---------------------------------------------------------------------------

def get_db_connection():
    """Create and return a new MySQL connection."""
    return mysql.connector.connect(**DB_CONFIG)


def ensure_table_exists(conn):
    """Create the asset_prices table if it doesn't exist."""
    ddl = """
    CREATE TABLE IF NOT EXISTS asset_prices (
        id          BIGINT       AUTO_INCREMENT PRIMARY KEY,
        asset_name  VARCHAR(50)  NOT NULL,
        symbol      VARCHAR(10)  NOT NULL,
        asset_type  ENUM('metal', 'crypto') NOT NULL,
        price_usd   DECIMAL(18, 8) NOT NULL,
        fetched_at  DATETIME     NOT NULL,
        created_at  DATETIME     DEFAULT CURRENT_TIMESTAMP,
        INDEX idx_symbol_time (symbol, fetched_at),
        INDEX idx_type_time   (asset_type, fetched_at)
    ) ENGINE=InnoDB;
    """
    with conn.cursor() as cur:
        cur.execute(ddl)
    conn.commit()
    logger.info("Ensured asset_prices table exists.")


def insert_prices(conn, records: list[dict]):
    """
    Insert a batch of price records into the database.

    Each record dict: {asset_name, symbol, asset_type, price_usd, fetched_at}
    """
    if not records:
        return

    sql = """
        INSERT INTO asset_prices (asset_name, symbol, asset_type, price_usd, fetched_at)
        VALUES (%(asset_name)s, %(symbol)s, %(asset_type)s, %(price_usd)s, %(fetched_at)s)
    """
    with conn.cursor() as cur:
        cur.executemany(sql, records)
    conn.commit()
    logger.info("Inserted %d price record(s) into database.", len(records))

# ---------------------------------------------------------------------------
# API fetch functions
# ---------------------------------------------------------------------------

def fetch_crypto_prices() -> list[dict]:
    """
    Fetch crypto prices from CoinGecko (free, no API key).

    Returns a list of record dicts ready for DB insertion.
    """
    ids = ",".join(CRYPTO_ASSETS.keys())
    params = {
        "ids": ids,
        "vs_currencies": "usd",
    }
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    response = requests.get(
        COINGECKO_URL,
        params=params,
        timeout=REQUEST_TIMEOUT,
        headers={"Accept": "application/json"},
    )
    response.raise_for_status()
    data = response.json()

    records = []
    for cg_id, (name, symbol) in CRYPTO_ASSETS.items():
        if cg_id in data and "usd" in data[cg_id]:
            price = Decimal(str(data[cg_id]["usd"]))
            records.append({
                "asset_name": name,
                "symbol":     symbol,
                "asset_type": "crypto",
                "price_usd":  price,
                "fetched_at": now,
            })
            logger.info("  %s (%s): $%s", name, symbol, price)
        else:
            logger.warning("  Missing data for %s in CoinGecko response.", cg_id)

    return records


def fetch_metal_prices() -> list[dict]:
    """
    Fetch precious metal prices from Metals.dev (free tier, API key required).

    Returns a list of record dicts ready for DB insertion.
    """
    if not METALS_API_KEY:
        logger.warning("METALS_API_KEY not set — skipping precious metals fetch. "
                       "Sign up free at https://metals.dev")
        return []

    params = {
        "api_key":  METALS_API_KEY,
        "currency": "USD",
        "unit":     "toz",  # troy ounce
    }
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    response = requests.get(
        METALS_URL,
        params=params,
        timeout=REQUEST_TIMEOUT,
        headers={"Accept": "application/json"},
    )
    response.raise_for_status()
    data = response.json()

    # metals.dev returns: {"metals": {"gold": 2345.67, "silver": 28.12, ...}}
    metals_data = data.get("metals", {})

    records = []
    for metal_key, (name, symbol) in METAL_ASSETS.items():
        if metal_key in metals_data:
            price = Decimal(str(metals_data[metal_key]))
            records.append({
                "asset_name": name,
                "symbol":     symbol,
                "asset_type": "metal",
                "price_usd":  price,
                "fetched_at": now,
            })
            logger.info("  %s (%s): $%s / troy oz", name, symbol, price)
        else:
            logger.warning("  Missing data for %s in metals.dev response.", metal_key)

    return records

# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------

def run_once(conn) -> int:
    """
    Perform a single fetch-and-store cycle.
    Returns the number of records inserted.
    """
    all_records = []

    # --- Fetch crypto prices ---
    logger.info("Fetching crypto prices from CoinGecko...")
    try:
        crypto_records = fetch_crypto_prices()
        all_records.extend(crypto_records)
    except requests.exceptions.Timeout:
        logger.error("CoinGecko request timed out.")
    except requests.exceptions.HTTPError as e:
        # Handle rate limiting specifically
        if e.response is not None and e.response.status_code == 429:
            logger.warning("CoinGecko rate limit hit (429). Will retry next cycle.")
        else:
            logger.error("CoinGecko HTTP error: %s", e)
    except requests.exceptions.ConnectionError:
        logger.error("CoinGecko connection failed — check your network.")
    except (ValueError, KeyError) as e:
        logger.error("Failed to parse CoinGecko response: %s", e)

    # --- Fetch metal prices ---
    logger.info("Fetching precious metal prices from Metals.dev...")
    try:
        metal_records = fetch_metal_prices()
        all_records.extend(metal_records)
    except requests.exceptions.Timeout:
        logger.error("Metals.dev request timed out.")
    except requests.exceptions.HTTPError as e:
        if e.response is not None and e.response.status_code == 429:
            logger.warning("Metals.dev rate limit hit (429). Will retry next cycle.")
        elif e.response is not None and e.response.status_code == 403:
            logger.error("Metals.dev returned 403 — check your API key.")
        else:
            logger.error("Metals.dev HTTP error: %s", e)
    except requests.exceptions.ConnectionError:
        logger.error("Metals.dev connection failed — check your network.")
    except (ValueError, KeyError) as e:
        logger.error("Failed to parse Metals.dev response: %s", e)

    # --- Store in DB ---
    if all_records:
        try:
            insert_prices(conn, all_records)
        except MySQLError as e:
            logger.error("MySQL insert failed: %s", e)
            # Attempt to reconnect on the next cycle
            raise

    return len(all_records)


def main():
    """Entry point — connects to MySQL and runs the polling loop forever."""
    logger.info("=" * 60)
    logger.info("Price Tracker starting up...")
    logger.info("Polling interval: %d seconds", POLL_INTERVAL)
    logger.info("=" * 60)

    # Initial DB connection
    try:
        conn = get_db_connection()
        ensure_table_exists(conn)
        logger.info("Connected to MySQL at %s:%s/%s",
                     DB_CONFIG["host"], DB_CONFIG["port"], DB_CONFIG["database"])
    except MySQLError as e:
        logger.critical("Cannot connect to MySQL: %s", e)
        logger.critical("Exiting. Fix your database config and retry.")
        sys.exit(1)

    cycle = 0
    while True:
        cycle += 1
        logger.info("--- Cycle %d ---", cycle)

        try:
            # Ensure connection is still alive; reconnect if needed
            if not conn.is_connected():
                logger.warning("MySQL connection lost — reconnecting...")
                conn = get_db_connection()
                logger.info("Reconnected to MySQL.")

            count = run_once(conn)
            logger.info("Cycle %d complete: %d records stored.", cycle, count)

        except MySQLError as e:
            logger.error("Database error in cycle %d: %s", cycle, e)
            # Try to get a fresh connection next cycle
            try:
                conn.close()
            except Exception:
                pass
            try:
                conn = get_db_connection()
                logger.info("Re-established MySQL connection.")
            except MySQLError:
                logger.error("Could not reconnect to MySQL. Will retry next cycle.")

        except Exception as e:
            # Catch-all so the loop never dies
            logger.exception("Unexpected error in cycle %d: %s", cycle, e)

        # Sleep until next cycle
        logger.info("Sleeping %d seconds...\n", POLL_INTERVAL)
        try:
            time.sleep(POLL_INTERVAL)
        except KeyboardInterrupt:
            logger.info("KeyboardInterrupt received — shutting down gracefully.")
            try:
                conn.close()
            except Exception:
                pass
            sys.exit(0)


if __name__ == "__main__":
    main()

---

How It Works — Architecture Overview

┌─────────────┐  every 60s   ┌──────────────────┐
│  main loop  │──────────────▶│   run_once()     │
└─────────────┘               │                  │
                              │  ┌─────────────┐ │     GET /simple/price
                              │  │fetch_crypto │─┼────▶ CoinGecko (free)
                              │  └─────────────┘ │
                              │  ┌─────────────┐ │     GET /v1/latest
                              │  │fetch_metals │─┼────▶ Metals.dev (free tier)
                              │  └─────────────┘ │
                              │         │         │
                              │    insert_prices()│
                              └─────────┼─────────┘
                                        │
                                        ▼
                              ┌──────────────────┐
                              │  MySQL           │
                              │  asset_prices    │
                              └──────────────────┘

---

Sample Output

2025-01-15 14:30:00 [INFO] --- Cycle 1 ---
2025-01-15 14:30:00 [INFO] Fetching crypto prices from CoinGecko...
2025-01-15 14:30:01 [INFO]   Bitcoin (BTC): $99432.00000000
2025-01-15 14:30:01 [INFO]   Ethereum (ETH): $3285.42000000
2025-01-15 14:30:01 [INFO]   BNB (BNB): $695.30000000
2025-01-15 14:30:01 [INFO]   Solana (SOL): $185.67000000
2025-01-15 14:30:01 [INFO]   XRP (XRP): $2.34000000
2025-01-15 14:30:01 [INFO] Fetching precious metal prices from Metals.dev...
2025-01-15 14:30:01 [INFO]   Gold (XAU): $2678.50000000 / troy oz
2025-01-15 14:30:01 [INFO]   Silver (XAG): $30.45000000 / troy oz
2025-01-15 14:30:01 [INFO]   Platinum (XPT): $952.00000000 / troy oz
2025-01-15 14:30:01 [INFO]   Palladium (XPD): $978.25000000 / troy oz
2025-01-15 14:30:01 [INFO] Inserted 9 price record(s) into database.
2025-01-15 14:30:01 [INFO] Cycle 1 complete: 9 records stored.
2025-01-15 14:30:01 [INFO] Sleeping 60 seconds...

---

Query Examples

-- Latest price for each asset
SELECT asset_name, symbol, price_usd, fetched_at
FROM asset_prices p1
INNER JOIN (
    SELECT symbol, MAX(fetched_at) AS max_ts
    FROM asset_prices
    GROUP BY symbol
) p2 ON p1.symbol = p2.symbol AND p1.fetched_at = p2.max_ts;

-- Gold price history for the last 24 hours
SELECT price_usd, fetched_at
FROM asset_prices
WHERE symbol = 'XAU'
  AND fetched_at >= NOW() - INTERVAL 24 HOUR
ORDER BY fetched_at;

-- Average Bitcoin price per hour today
SELECT DATE_FORMAT(fetched_at, '%Y-%m-%d %H:00') AS hour,
       AVG(price_usd) AS avg_price,
       MIN(price_usd) AS low,
       MAX(price_usd) AS high
FROM asset_prices
WHERE symbol = 'BTC' AND DATE(fetched_at) = CURDATE()
GROUP BY hour
ORDER BY hour;

---

Key Design Decisions

Rate Limit Budget

⚠️ Important: At 1 request/minute, you'll exceed Metals.dev's 50/day free tier. Solutions:

1. Poll metals less frequently — change the metals fetch to every 30 minutes (metals don't move as fast as crypto). Add a simple cycle % 30 == 0 check.
2. Use GoldAPI.io instead — 300 requests/month free (works for ~every-2-hour polling).
3. Upgrade to Metals.dev paid tier ($9/mo for 5,000 req/day).

To poll metals every 30 minutes instead, change run_once() to:

# Only fetch metals every 30 cycles (30 minutes)
if cycle % 30 == 1:
    logger.info("Fetching precious metal prices from Metals.dev...")
    # ... fetch metals ...
else:
    logger.info("Skipping metals fetch (next in %d min).", 30 - (cycle % 30))