""" BullionVault Gold Price Service Fetches real-time gold prices from BullionVault's CSV data API """ from __future__ import annotations import asyncio from datetime import datetime, timezone import logging from typing import Optional, Dict, Any, List import csv import io import httpx logger = logging.getLogger(__name__) class BullionVaultService: """ Service to fetch gold prices from BullionVault BullionVault provides accurate, real-time precious metals prices Uses their CSV data API: https://chart-data.bullionvault.com """ # Correct BullionVault CSV API base URL BASE_URL = "https://chart-data.bullionvault.com" # Metal codes METALS = { 'gold': 'AUX', 'silver': 'AGX', 'platinum': 'PTX', 'palladium': 'PDX' } # Interval codes (seconds between data points) INTERVALS = { '10m': 5, # 10 minutes '1h': 15, # 1 hour '6h': 120, # 6 hours '1d': 600, # 1 day (default) '1w': 3600, # 1 week '1m': 14400, # 1 month '3m': 43200, # 3 months (1 quarter) '1y': 172800, # 1 year '5y': 864000, # 5 years '20y': 2592000 # 20 years } def __init__( self, client: Optional[httpx.AsyncClient] = None, *, base_url: Optional[str] = None, timeout: float = 30.0, max_retries: int = 3, retry_backoff_seconds: float = 0.5, ) -> None: self.base_url = base_url or self.BASE_URL self.max_retries = max(1, max_retries) self.retry_backoff_seconds = max(0.0, retry_backoff_seconds) if client is None: self.client = httpx.AsyncClient(base_url=self.base_url, timeout=timeout) self._owns_client = True else: self.client = client self._owns_client = False async def __aenter__(self) -> "BullionVaultService": return self async def __aexit__(self, *exc_info: object) -> None: await self.close() 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 CSV data from BullionVault # Format: /prices/CSV/{metal}/{currency}/{interval}/Full metal_code = self.METALS['gold'] interval = self.INTERVALS['1d'] path = f"/prices/CSV/{metal_code}/{currency.upper()}/{interval}/Full" csv_text = await self._fetch_csv(path) # Parse CSV data price_data = self._parse_csv(csv_text) if not price_data: raise ValueError("No price data available from BullionVault") # Get latest price (first row after header) latest = price_data[0] # Calculate daily statistics oz_prices = [row['oz_close'] for row in price_data if row['oz_close'] is not None] if not oz_prices: raise ValueError("No valid price points") current_price = latest['oz_close'] daily_high = max([row['oz_high'] for row in price_data if row['oz_high'] is not None]) daily_low = min([row['oz_low'] for row in price_data if row['oz_low'] is not None]) # Calculate change from last data point first_price = price_data[-1]['oz_close'] if len(price_data) > 1 else current_price change = current_price - first_price change_percent = (change / first_price * 100) if first_price else 0.0 timestamp = latest['timestamp'] result = { "price": round(current_price, 2), "price_kg": round(latest['kg_close'], 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(), "source": "BullionVault", "trading_day": timestamp.strftime("%Y-%m-%d"), "data_points": len(price_data) } 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 _fetch_csv(self, path: str) -> str: """Fetch CSV data from BullionVault with simple retry logic.""" last_exception: Optional[Exception] = None base = self.base_url.rstrip("/") for attempt in range(1, self.max_retries + 1): try: url = path if path.startswith("http") else f"{base}{path}" response = await self.client.get(url) response.raise_for_status() csv_text = response.text.strip() if not csv_text: raise ValueError("BullionVault returned empty response body") logger.debug( "Fetched BullionVault CSV successfully", extra={"path": url, "attempt": attempt}, ) return csv_text except (httpx.RequestError, httpx.HTTPStatusError, ValueError) as exc: last_exception = exc logger.warning( "BullionVault CSV fetch attempt failed", extra={ "path": url if "url" in locals() else path, "attempt": attempt, "max_attempts": self.max_retries, "error": str(exc), }, ) if attempt < self.max_retries: await asyncio.sleep(self.retry_backoff_seconds * attempt) assert last_exception is not None raise last_exception def _parse_csv(self, csv_text: str) -> List[Dict[str, Any]]: """ Parse BullionVault CSV response CSV format: "Date",High (kg),Low (kg),Close (kg),,High (troy oz),Low (troy oz),Close (troy oz), "05:10:00 23-Nov-2025",130702.99,130702.99,130702.99,,4065.32,4065.32,4065.32, Args: csv_text: Raw CSV text from BullionVault Returns: List of price dictionaries """ result = [] # Parse CSV reader = csv.reader(io.StringIO(csv_text)) # Skip header next(reader, None) for row in reader: if len(row) < 8: continue try: # Parse date/time: "HH:MM:SS DD-Mon-YYYY" date_str = row[0].strip('"') timestamp = datetime.strptime(date_str, "%H:%M:%S %d-%b-%Y").replace(tzinfo=timezone.utc) # Extract prices (kg and oz) kg_high = self._to_float(row[1]) kg_low = self._to_float(row[2]) kg_close = self._to_float(row[3]) oz_high = self._to_float(row[5]) oz_low = self._to_float(row[6]) oz_close = self._to_float(row[7]) result.append({ 'timestamp': timestamp, 'kg_high': kg_high, 'kg_low': kg_low, 'kg_close': kg_close, 'oz_high': oz_high, 'oz_low': oz_low, 'oz_close': oz_close }) except (ValueError, IndexError) as e: logger.warning(f"Skipping malformed CSV row: {row} - {e}") continue result.sort(key=lambda entry: entry['timestamp'], reverse=True) return result @staticmethod def _to_float(value: Optional[str]) -> Optional[float]: if value in (None, ""): return None try: return float(value) except (TypeError, ValueError): return None 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, 3m, 1y, 5y, 20y) limit: Maximum number of data points to return Returns: List of OHLC data points """ try: metal_code = self.METALS['gold'] interval = self.INTERVALS.get(timeframe, self.INTERVALS['1d']) path = f"/prices/CSV/{metal_code}/{currency.upper()}/{interval}/Full" csv_text = await self._fetch_csv(path) # Parse CSV data price_data = self._parse_csv(csv_text) # Apply limit if specified if limit and len(price_data) > limit: price_data = price_data[:limit] # Convert to OHLCV format result = [] for point in price_data: result.append({ "timestamp": point['timestamp'].isoformat(), "time": int(point['timestamp'].timestamp()), "open": point['oz_close'], # BullionVault doesn't provide open, use close "high": point['oz_high'], "low": point['oz_low'], "close": point['oz_close'], "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""" if self._owns_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)