"""Helper functions to fetch macro data and global markets.""" import logging from datetime import datetime, timedelta from typing import Dict, Optional, Any import pandas as pd import yfinance as yf from ..common.logging import logger def fetch_global_market_data(tickers: Optional[list] = None) -> Dict[str, Any]: if tickers is None: tickers = [ "^GSPC", # S&P 500 "^DJI", # Dow Jones "^IXIC", # Nasdaq "GC=F", # Gold "SI=F", # Silver "CL=F", # WTI Crude "BTC-USD", # Bitcoin ] result = {} for ticker in tickers: try: data = yf.download(ticker, period="1mo", interval="1d", progress=False) if data.empty: continue current = data['Close'].iloc[-1] month_ago = data['Close'].iloc[0] if len(data) > 0 else current # Ensure scalar values current_val = float(current.iloc[0]) if hasattr(current, 'iloc') else float(current) month_val = float(month_ago.iloc[0]) if hasattr(month_ago, 'iloc') else float(month_ago) month_change = (current_val - month_val) / month_val * 100 result[ticker] = { "current": current_val, "month_change_pct": month_change } except Exception as e: logger.error(f"Error fetching {ticker}: {e}") return result def prepare_macro_data(scraper) -> Dict[str, Any]: logger.info("Fetching macro data...") inflation = scraper.fetch_inflation() interest_rates = scraper.fetch_interest_rates() nrb = scraper.fetch_nrb_data() forex = scraper.fetch_forex_rates() gdp = scraper.fetch_gdp_vs_nepse() global_data = fetch_global_market_data() return { "inflation_data": inflation, "interest_rates_data": interest_rates, "nrb_data": nrb, "forex_data": forex, "gdp_data": gdp, "global_data": global_data, }