update_crypto_prices.py

#!/usr/bin/env python3
"""
Direct database updater for crypto prices
Fetches latest prices from CoinMarketCap and CoinGecko
and updates the database directly
"""

import argparse
import logging
import os
import sys
import time
from datetime import datetime

import requests
from dotenv import load_dotenv

# Add project to path
project_root_path = os.path.abspath(os.path.dirname(__file__))
sys.path.append(project_root_path)

# Moved imports to top (E402 fix)
from src.database.client import DatabaseClient

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",  # Line break (E501 fix)
)
logger = logging.getLogger(__name__)

# CoinMarketCap API Configuration
CMC_API_KEY = os.getenv(
    "CMC_API_KEY", "6e1f5ef3-47fc-4288-9b3e-f02a7493be6a"
)  # Default to provided key
CMC_BASE_URL = "https://pro-api.coinmarketcap.com/v1"

# CoinGecko API Configuration
COINGECKO_BASE_URL = "https://api.coingecko.com/api/v3"

# Asset mapping between database ID, symbol, and CoinGecko ID
ASSET_MAPPING = [
    {"db_id": 1, "symbol": "BTC", "coingecko_id": "bitcoin"},
    {"db_id": 2, "symbol": "ETH", "coingecko_id": "ethereum"},
    {"db_id": 3, "symbol": "SOL", "coingecko_id": "solana"},
    {"db_id": 4, "symbol": "BNB", "coingecko_id": "binancecoin"},
    {"db_id": 5, "symbol": "XRP", "coingecko_id": "ripple"},
]


def fetch_coinmarketcap_data(symbol="BTC"):
    """Fetch cryptocurrency data from CoinMarketCap API."""
    logger.info(f"Fetching {symbol} data from CoinMarketCap...")

    url = f"{CMC_BASE_URL}/cryptocurrency/quotes/latest"
    headers = {"X-CMC_PRO_API_KEY": CMC_API_KEY, "Accept": "application/json"}
    params = {"symbol": symbol, "convert": "USD"}

    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()

        if "data" in data and symbol in data["data"]:
            coin_data = data["data"][symbol]
            quote = coin_data["quote"]["USD"]

            result = {
                "source": "coinmarketcap",
                "timestamp": datetime.now().isoformat(),
                "symbol": symbol,
                "name": coin_data["name"],
                "price_usd": quote["price"],
                "market_cap_usd": quote["market_cap"],
                "volume_24h_usd": quote["volume_24h"],
                "price_change_24h_pct": quote["percent_change_24h"],
                "additional_data": {
                    "circulating_supply": coin_data["circulating_supply"],
                    "total_supply": coin_data["total_supply"],
                    "max_supply": coin_data["max_supply"],
                    "price_change_1h": quote["percent_change_1h"],
                    "price_change_7d": quote["percent_change_7d"],
                    "price_change_30d": quote["percent_change_30d"],
                },
            }

            return result
        else:
            logger.error(f"Unexpected response structure from CMC: {data}")
            return None

    except requests.exceptions.RequestException as e:
        logger.error(f"Error fetching data from CMC: {e}")
        if hasattr(e, "response") and hasattr(e.response, "text"):
            logger.error(f"Response: {e.response.text}")  # Line break (E501 fix)
        return None


def fetch_coingecko_data(coin_id="bitcoin"):
    """Fetch cryptocurrency data from CoinGecko API."""
    logger.info(f"Fetching {coin_id} data from CoinGecko...")

    url = f"{COINGECKO_BASE_URL}/coins/{coin_id}"
    params = {
        "localization": "false",
        "tickers": "false",
        "market_data": "true",
        "community_data": "false",
        "developer_data": "false",
    }

    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()

        if "market_data" in data:
            market_data = data["market_data"]

            result = {
                "source": "coingecko",
                "timestamp": datetime.now().isoformat(),
                "symbol": data["symbol"].upper(),
                "name": data["name"],
                "price_usd": market_data["current_price"]["usd"],
                "market_cap_usd": market_data["market_cap"]["usd"],
                "volume_24h_usd": market_data["total_volume"]["usd"],
                "price_change_24h_pct": market_data["price_change_percentage_24h"],
                "additional_data": {
                    "circulating_supply": market_data.get("circulating_supply"),
                    "total_supply": market_data.get("total_supply"),
                    "max_supply": market_data.get("max_supply"),
                    "price_change_1h": market_data.get(
                        "price_change_percentage_1h_in_currency", {}
                    ).get(
                        "usd"
                    ),  # Line break (E501 fix)
                    "price_change_7d": market_data.get(
                        "price_change_percentage_7d"
                    ),  # Line break (E501 fix)
                    "price_change_30d": market_data.get(
                        "price_change_percentage_30d"
                    ),  # Line break (E501 fix)
                    "ath": market_data.get("ath", {}).get("usd"),
                    "low_24h": market_data.get("low_24h", {}).get("usd"),
                    "high_24h": market_data.get("high_24h", {}).get("usd"),
                },
            }

            return result
        else:
            logger.error(f"Unexpected response structure from CG: {data}")
            return None

    except requests.exceptions.RequestException as e:
        logger.error(f"Error fetching data from CG: {e}")
        if hasattr(e, "response") and hasattr(e.response, "text"):
            logger.error(f"Response: {e.response.text}")  # Line break (E501 fix)
        return None


def update_database(db_client, asset_id, data, primary_source="coinmarketcap"):
    """
    Update the database with the latest cryptocurrency data.

    Args:
        db_client: DatabaseClient instance
        asset_id: Database ID for the asset
        data: Dictionary containing data from both CoinMarketCap and CoinGecko
        primary_source: Which source to prioritize ('coinmarketcap' or 'coingecko')
    """
    if not data.get(primary_source):
        logger.warning(f"No {primary_source} data for asset_id={asset_id}")
        # Try fallback source
        fallback = (
            "coingecko" if primary_source == "coinmarketcap" else "coinmarketcap"
        )  # Line break (E501 fix)
        if not data.get(fallback):
            logger.error(f"No data from either source for asset_id={asset_id}")
            return False
        primary_source = fallback

    # Get data from the selected source
    source_data = data[primary_source]

    # Create a record to insert into the database
    record = {
        "asset_id": asset_id,
        "timestamp": source_data["timestamp"],
        "price_usd": source_data["price_usd"],
        "market_cap_usd": source_data["market_cap_usd"],
        "volume_24h_usd": source_data["volume_24h_usd"],
        "price_change_24h_pct": source_data["price_change_24h_pct"],
        "source": source_data["source"],
        "additional_data": source_data["additional_data"],
    }

    # Insert the record into the database
    logger.info(
        f"Inserting record for asset_id={asset_id} " f"from source={primary_source}"
    )
    try:
        # Use the database client to insert the record
        response = db_client.client.table("market_data").insert(record).execute()
        if hasattr(response, "data") and response.data:
            logger.info(f"Successfully inserted record for asset_id={asset_id}")
            return True
        else:
            logger.warning(
                f"Insertion for asset_id={asset_id} returned no data"
            )  # Line break (E501 fix)
            return False
    except Exception as e:
        logger.error(f"Failed to insert record for asset_id={asset_id}: {e}")
        return False


def fetch_and_update_asset(db_client, asset, primary_source="coinmarketcap"):
    """Fetch data for a single asset and update the database."""
    symbol = asset["symbol"]
    coingecko_id = asset["coingecko_id"]
    asset_id = asset["db_id"]

    # Fetch data from both sources
    data = {}

    # Try CoinMarketCap
    try:
        data["coinmarketcap"] = fetch_coinmarketcap_data(symbol)
    except Exception as e:
        logger.error(f"Failed to fetch {symbol} from CoinMarketCap: {e}")

    # Add delay to avoid rate limits
    time.sleep(1)

    # Try CoinGecko
    try:
        data["coingecko"] = fetch_coingecko_data(coingecko_id)
    except Exception as e:
        logger.error(f"Failed to fetch {coingecko_id} from CoinGecko: {e}")

    # Update database with the data
    success = update_database(db_client, asset_id, data, primary_source)

    # Print a comparison if both sources were successful
    if data.get("coinmarketcap") and data.get("coingecko"):
        cmc_price = data["coinmarketcap"]["price_usd"]
        cg_price = data["coingecko"]["price_usd"]
        diff = ((cg_price - cmc_price) / cmc_price) * 100 if cmc_price else 0

        logger.info(
            f"{symbol} Price Comparison: CMC=${cmc_price:.2f}, "
            f"CG=${cg_price:.2f} (Diff: {diff:.2f}%)"
        )

    return success


def update_all_assets(primary_source="coinmarketcap"):
    """Update all assets in the database with latest prices."""
    # Initialize database client
    logger.info("Initializing database client...")
    db_client = DatabaseClient()

    if not db_client.client:
        logger.error("Failed to initialize database client!")
        return False

    logger.info("Database client initialized successfully.")

    # Get list of assets from the database to ensure we only update active ones
    try:
        db_assets = db_client.get_assets()
        if not db_assets:
            logger.warning("No assets found in the database!")
            return False

        # Create a mapping of assets to update
        assets_to_update = []
        for mapping in ASSET_MAPPING:
            for db_asset in db_assets:
                if db_asset["symbol"] == mapping["symbol"]:
                    assets_to_update.append(
                        {
                            "db_id": db_asset["id"],
                            "symbol": mapping["symbol"],
                            "coingecko_id": mapping["coingecko_id"],
                        }
                    )
                    break

        if not assets_to_update:
            logger.warning("No matching assets to update!")
            return False

        # Update each asset
        success_count = 0
        for asset in assets_to_update:
            logger.info(f"Processing {asset['symbol']}...")
            if fetch_and_update_asset(db_client, asset, primary_source):
                success_count += 1
            # Add delay to avoid rate limits
            time.sleep(2)

        logger.info(
            f"Successfully updated {success_count}/{len(assets_to_update)} assets."
        )
        return success_count > 0

    except Exception as e:
        logger.error(f"Error updating assets: {e}")
        return False


def main():
    """Run the price updater."""
    parser = argparse.ArgumentParser(description="Update crypto prices in the database")
    parser.add_argument(
        "--source",
        choices=["coinmarketcap", "coingecko"],
        default="coinmarketcap",
        help="Primary data source",
    )
    parser.add_argument("--debug", action="store_true", help="Enable debug logging")

    args = parser.parse_args()

    # Set logging level
    if args.debug:
        logging.getLogger().setLevel(logging.DEBUG)

    logger.info(
        f"Starting crypto price updater using {args.source} as " f"primary source..."
    )
    success = update_all_assets(primary_source=args.source)

    if success:
        logger.info("Price update completed successfully.")
        return 0
    else:
        logger.error("Price update failed!")
        return 1


if __name__ == "__main__":
    sys.exit(main())