import logging import yfinance as yf from decimal import Decimal from datetime import datetime from typing import List, Dict, Any logger = logging.getLogger(__name__) # Ticker mapping # HO=F : Heating Oil # CL=F : Crude Oil # RB=F : RBOB Gasoline TICKERS = ["HO=F", "CL=F", "RB=F"] def fetch_ticker_data() -> List[Dict[str, Any]]: """ Fetch current data for oil tickers from Yahoo Finance. Returns: List of dictionaries containing ticker data. """ results = [] try: # Fetch data for all tickers at once tickers_str = " ".join(TICKERS) data = yf.Tickers(tickers_str) for symbol in TICKERS: try: ticker = data.tickers[symbol] # Fast info usually contains the latest price info = ticker.fast_info # Fallback to history if fast_info is missing crucial data (simplified here) # But fast_info is generally faster and sufficient for last_price last_price = info.last_price previous_close = info.previous_close if last_price is None: logger.warning(f"No price found for {symbol}") continue change = last_price - previous_close percent_change = (change / previous_close) * 100 if previous_close else 0 results.append({ "symbol": symbol, "price": Decimal(str(last_price)), "currency": info.currency, "change": Decimal(str(change)), "percent_change": Decimal(str(percent_change)), "timestamp": datetime.utcnow() }) logger.info(f"Fetched {symbol}: {last_price} ({percent_change:.2f}%)") except Exception as e: logger.error(f"Error processing {symbol}: {e}") except Exception as e: logger.error(f"Error fetching ticker data: {e}") return results