Write a Python script that continuously fetches the following prices every 60 seconds and stores them in a MySQL database: **Precious Metals:** Gold, Silver, Platinum, Palladium **Crypto:** Bitcoin, Ethereum, BNB, Solana, XRP (current top 5 by market cap) Requirements: - Use a free, publicly available API (no paid API key required, or at minimum a free tier with generous limits). Clearly state which API(s) you're using and any signup steps needed. - Fetch all prices in USD. - Store results in a MySQL database with appropriate table schema(s). Include the SQL to create the table(s). - Each record should include the asset name/symbol, price, and timestamp. - The script should run indefinitely, polling every 60 seconds. - Handle common errors gracefully (API timeouts, connection failures, rate limits) — log errors but don't crash. - Use clear, well-structured code with comments. - List all required Python packages (e.g., `requests`, `mysql-connector-python`, etc.) and how to install them.
Of course! Here is a complete solution to create a Python script that fetches and stores precious metal and cryptocurrency prices in a MySQL database.
This response is broken down into the following sections:
- API Selection & Setup: Which APIs we'll use and how to get started.
- Prerequisites: Python packages you'll need to install.
- MySQL Database Setup: The SQL command to create the necessary table.
- The Python Script: The full, commented code.
- How to Run the Script: Final instructions.
1. API Selection & Setup
To meet the requirements, we'll use two separate, highly reliable APIs with generous free tiers.
A) For Cryptocurrency Prices: CoinGecko CoinGecko offers a free, public API that does not require an API key for simple price lookups. It's fast, reliable, and has a very generous rate limit (50 calls/minute), which is perfect for our needs.
- Signup: No signup or API key is needed for this script.
- Endpoint we'll use:
https://api.coingecko.com/api/v3/simple/price
B) For Precious Metal Prices: Metals-API.com
Finding a free, high-frequency API for precious metals is challenging. We'll use metals-api.com, which offers a free plan that is great for development and testing.
- Signup: You need to sign up for a free account to get an API key.
- Go to https://metals-api.com/
- Click on "Get Free API Key" and register.
- You will receive a free API key.
- Important Note on Free Tier Limits: The free plan is limited to 50 API calls per month. For a script running every 60 seconds, you would need a paid plan. This script is built to be production-ready, but to run it continuously, you will need to upgrade your
metals-api.complan. For testing, it will work perfectly.
2. Prerequisites
You need to install three Python packages. You can install them all with a single command in your terminal:
pip install requests mysql-connector-python
requests: For making HTTP requests to the APIs.mysql-connector-python: The official Oracle driver for connecting Python to a MySQL database.
3. MySQL Database Setup
Before running the script, you need to have a MySQL server running and a database created. Connect to your MySQL server and run the following SQL command to create the asset_prices table.
CREATE TABLE asset_prices (
id INT AUTO_INCREMENT PRIMARY KEY,
asset_symbol VARCHAR(10) NOT NULL,
asset_type ENUM('crypto', 'metal') NOT NULL,
price_usd DECIMAL(20, 8) NOT NULL,
timestamp DATETIME NOT NULL,
INDEX (asset_symbol, timestamp)
);
Schema Explanation:
id: A unique identifier for each price record.asset_symbol: The symbol for the asset (e.g., 'BTC', 'XAU' for Gold).asset_type: AnENUMto easily distinguish between crypto and metals.price_usd: ADECIMALtype is used for financial data to avoid floating-point inaccuracies.(20, 8)allows for very large numbers and high precision.timestamp: The date and time when the price was fetched.INDEX: An index onasset_symbolandtimestampwill speed up queries if you later want to analyze the price history of a specific asset.
4. The Python Script
Here is the complete, well-commented Python script. Save it as price_tracker.py.
Important: Be sure to replace the placeholder values in the ## --- CONFIGURATION --- section with your actual database credentials and API key. For better security, consider using environment variables instead of hardcoding credentials.
import requests
import mysql.connector
import time
import datetime
import logging
## --- CONFIGURATION ---
# MySQL Database Configuration
DB_CONFIG = {
'host': 'localhost',
'user': 'your_db_user',
'password': 'your_db_password',
'database': 'your_db_name'
}
# Metals-API.com Configuration
METALS_API_KEY = 'YOUR_METALS_API_KEY' # Replace with your key
METALS_API_URL = f"https://metals-api.com/api/latest?access_key={METALS_API_KEY}"
# CoinGecko API Configuration
CRYPTO_API_URL = "https://api.coingecko.com/api/v3/simple/price"
# Assets to track
METAL_SYMBOLS = ['XAU', 'XAG', 'XPT', 'XPD'] # Gold, Silver, Platinum, Palladium
CRYPTO_IDS = ['bitcoin', 'ethereum', 'binancecoin', 'solana', 'ripple'] # BTC, ETH, BNB, SOL, XRP
# Script settings
POLL_INTERVAL_SECONDS = 60
## --- SCRIPT SETUP ---
# Configure logging to show errors and info
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
## --- FUNCTIONS ---
def get_db_connection():
"""Establishes and returns a connection to the MySQL database."""
try:
conn = mysql.connector.connect(**DB_CONFIG)
logging.info("Successfully connected to the database.")
return conn
except mysql.connector.Error as err:
logging.error(f"Error connecting to database: {err}")
return None
def fetch_metal_prices():
"""Fetches precious metal prices from Metals-API.com."""
params = {
'base': 'USD',
'symbols': ','.join(METAL_SYMBOLS)
}
try:
response = requests.get(METALS_API_URL, params=params, timeout=10)
response.raise_for_status() # Raises an HTTPError for bad responses (4xx or 5xx)
data = response.json()
if not data.get('success'):
error_info = data.get('error', {})
logging.error(f"Metals API Error: {error_info.get('type')} - {error_info.get('info')}")
return {}
# Prices are per ounce, convert them to a standard format
prices = {symbol: 1 / data['rates'][symbol] for symbol in METAL_SYMBOLS}
logging.info(f"Successfully fetched metal prices: {prices}")
return prices
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching metal prices: {e}")
return {}
def fetch_crypto_prices():
"""Fetches cryptocurrency prices from CoinGecko."""
params = {
'ids': ','.join(CRYPTO_IDS),
'vs_currencies': 'usd'
}
try:
response = requests.get(CRYPTO_API_URL, params=params, timeout=10)
response.raise_for_status()
data = response.json()
# Extract prices and format them with uppercase symbols
# e.g., {'bitcoin': {'usd': 60000}} -> {'BTC': 60000}
symbol_map = {
'bitcoin': 'BTC', 'ethereum': 'ETH', 'binancecoin': 'BNB',
'solana': 'SOL', 'ripple': 'XRP'
}
prices = {symbol_map[id]: data[id]['usd'] for id in CRYPTO_IDS if id in data}
logging.info(f"Successfully fetched crypto prices: {prices}")
return prices
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching crypto prices: {e}")
return {}
def insert_prices_to_db(conn, prices_data):
"""Inserts a list of price data into the database."""
if not prices_data or not conn:
return
sql = """
INSERT INTO asset_prices (asset_symbol, asset_type, price_usd, timestamp)
VALUES (%s, %s, %s, %s)
"""
cursor = conn.cursor()
try:
cursor.executemany(sql, prices_data)
conn.commit()
logging.info(f"Successfully inserted {cursor.rowcount} records into the database.")
except mysql.connector.Error as err:
logging.error(f"Database insert error: {err}")
conn.rollback()
finally:
cursor.close()
## --- MAIN EXECUTION ---
def main():
"""Main loop to fetch and store prices indefinitely."""
while True:
logging.info("--- Starting new fetch cycle ---")
# Get current timestamp for this batch of prices
fetch_timestamp = datetime.datetime.now()
# Fetch prices from all sources
metal_prices = fetch_metal_prices()
crypto_prices = fetch_crypto_prices()
# Prepare data for database insertion
records_to_insert = []
for symbol, price in metal_prices.items():
records_to_insert.append((symbol, 'metal', price, fetch_timestamp))
for symbol, price in crypto_prices.items():
records_to_insert.append((symbol, 'crypto', price, fetch_timestamp))
# Store results in the database
if records_to_insert:
db_conn = get_db_connection()
if db_conn:
insert_prices_to_db(db_conn, records_to_insert)
db_conn.close()
logging.info("Database connection closed.")
else:
logging.warning("No price data was fetched in this cycle. Nothing to insert.")
# Wait for the next interval
logging.info(f"--- Cycle finished. Waiting for {POLL_INTERVAL_SECONDS} seconds... ---")
time.sleep(POLL_INTERVAL_SECONDS)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
logging.info("Script terminated by user.")
5. How to Run the Script
- Save the Code: Save the script above as
price_tracker.py. - Fill in Credentials: Open the file and replace the placeholder values in the
DB_CONFIGdictionary and theMETALS_API_KEYvariable with your own. - Run from Terminal: Open your terminal or command prompt, navigate to the directory where you saved the file, and run the script:
python price_tracker.py
The script will now start running. You will see log messages in your terminal every 60 seconds indicating the status of the API fetches and database insertions. To stop the script, press Ctrl+C.
Comments