""" BullionVault Gold Price Service Fetches real-time gold prices from BullionVault's chart data API """ from __future__ import annotations import asyncio import httpx from typing import Optional, Dict, Any, List from datetime import datetime import logging import json logger = logging.getLogger(__name__) class BullionVaultService: """ Service to fetch gold prices from BullionVault BullionVault provides accurate, real-time precious metals prices """ def __init__(self): self.client = httpx.AsyncClient(timeout=15.0) self.base_url = "https://www.bullionvault.com" # BullionVault chart data endpoint self.chart_data_url = f"{self.base_url}/chart/chart-data.json" async def get_current_gold_price(self, currency: str = "USD") -> Dict[str, Any]: """ Get current gold spot price from BullionVault Args: currency: Currency code (USD, GBP, EUR, JPY, AUD, CAD, CHF) Returns: Dict with price, high, low, change, timestamp, etc. """ try: # Fetch latest gold price data params = { "bullion": "gold", "currency": currency.upper(), "timeframe": "1d", # 1 day for recent data "chartType": "line" } response = await self.client.get(self.chart_data_url, params=params) response.raise_for_status() data = response.json() if not data or "prices" not in data: raise ValueError("Invalid response from BullionVault") prices = data["prices"] if not prices: raise ValueError("No price data available") # Get latest price point latest = prices[-1] # Calculate daily statistics daily_prices = [p[1] for p in prices if p[1] is not None] if not daily_prices: raise ValueError("No valid price points") current_price = latest[1] # Price per ounce daily_high = max(daily_prices) daily_low = min(daily_prices) # Calculate change from first price of day first_price = prices[0][1] change = current_price - first_price change_percent = (change / first_price * 100) if first_price else 0.0 # Convert timestamp (BullionVault uses milliseconds) timestamp_ms = latest[0] timestamp = datetime.fromtimestamp(timestamp_ms / 1000.0) result = { "price": round(current_price, 2), "open": round(first_price, 2), "high": round(daily_high, 2), "low": round(daily_low, 2), "previous_close": round(first_price, 2), "change": round(change, 2), "change_percent": round(change_percent, 4), "currency": currency.upper(), "unit": "per troy oz", "timestamp": timestamp.isoformat(), "timestamp_ms": timestamp_ms, "source": "BullionVault", "trading_day": timestamp.strftime("%Y-%m-%d"), "data_points": len(prices) } logger.info(f"✅ BullionVault gold price: {currency} ${current_price:.2f}/oz") return result except httpx.HTTPError as e: logger.error(f"❌ BullionVault HTTP error: {e}") raise except Exception as e: logger.error(f"❌ BullionVault price fetch failed: {e}") raise async def get_gold_history( self, currency: str = "USD", timeframe: str = "1d", limit: Optional[int] = None ) -> List[Dict[str, Any]]: """ Get historical gold price data from BullionVault Args: currency: Currency code timeframe: Time range (10m, 1h, 6h, 1d, 1w, 1m, 1q, 1y, 5y, 20y) limit: Maximum number of data points to return Returns: List of OHLC data points """ try: params = { "bullion": "gold", "currency": currency.upper(), "timeframe": timeframe, "chartType": "hlc" # High-Low-Close for OHLC data } response = await self.client.get(self.chart_data_url, params=params) response.raise_for_status() data = response.json() if not data or "prices" not in data: return [] prices = data["prices"] # Apply limit if specified if limit and len(prices) > limit: prices = prices[-limit:] # Convert to OHLCV format result = [] for point in prices: if len(point) >= 4: # [timestamp, open, high, low, close] timestamp_ms = point[0] result.append({ "timestamp": datetime.fromtimestamp(timestamp_ms / 1000.0).isoformat(), "time": int(timestamp_ms / 1000), "open": float(point[1]) if point[1] is not None else 0.0, "high": float(point[2]) if point[2] is not None else 0.0, "low": float(point[3]) if point[3] is not None else 0.0, "close": float(point[4]) if len(point) > 4 and point[4] is not None else float(point[1]), "volume": 0, # BullionVault doesn't provide volume }) logger.info(f"✅ BullionVault history: {len(result)} points for {timeframe}") return result except Exception as e: logger.error(f"❌ BullionVault history fetch failed: {e}") return [] async def get_multi_currency_prices(self) -> Dict[str, Dict[str, Any]]: """ Get current gold prices in multiple currencies Returns: Dict mapping currency codes to price data """ currencies = ["USD", "GBP", "EUR", "JPY", "AUD", "CAD", "CHF"] tasks = [self.get_current_gold_price(curr) for curr in currencies] results = await asyncio.gather(*tasks, return_exceptions=True) prices = {} for curr, result in zip(currencies, results): if isinstance(result, dict): prices[curr] = result else: logger.warning(f"Failed to fetch {curr} price: {result}") return prices async def close(self): """Close HTTP client""" await self.client.aclose() # Global instance bullionvault_service = BullionVaultService() # Convenience functions async def get_bullionvault_gold_price(currency: str = "USD") -> Dict[str, Any]: """Get current gold price from BullionVault""" return await bullionvault_service.get_current_gold_price(currency) async def get_bullionvault_history( currency: str = "USD", timeframe: str = "1d", limit: Optional[int] = None ) -> List[Dict[str, Any]]: """Get historical gold prices from BullionVault""" return await bullionvault_service.get_gold_history(currency, timeframe, limit)