Spaces:
Runtime error
Runtime error
| """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, | |
| } |