import yfinance as yf import pandas as pd from typing import List from .deriv_fetcher import DerivDataFetcher class DataFetcher: """Unified data fetcher for forex, indices, and synthetic volatility products.""" YAHOO_PAIRS = { # Major Forex "EUR/USD": "EURUSD=X", "GBP/USD": "GBPUSD=X", "USD/JPY": "USDJPY=X", "USD/CHF": "USDCHF=X", "AUD/USD": "AUDUSD=X", "USD/CAD": "USDCAD=X", "NZD/USD": "NZDUSD=X", "EUR/GBP": "EURGBP=X", "GBP/JPY": "GBPJPY=X", "EUR/JPY": "EURJPY=X", # Indices "S&P 500": "^GSPC", "NASDAQ": "^IXIC", "Dow Jones": "^DJI", "DAX": "^GDAXI", "FTSE 100": "^FTSE", "Nikkei 225": "^N225", # Volatility Indices (CBOE) "Volatility S&P 500 (VIX)": "^VIX", "Volatility NASDAQ 100 (VXN)": "^VXN", "Volatility Dow Jones (VXD)": "^VXD", "Volatility Russell 2000 (RVX)": "^RVX", "Volatility Crude Oil (OVX)": "^OVX", "Volatility Gold (GVZ)": "^GVZ", "Volatility EuroCurrency (EVZ)": "^EVZ", } _deriv_fetcher = None @classmethod def _get_deriv_fetcher(cls): if cls._deriv_fetcher is None: cls._deriv_fetcher = DerivDataFetcher() return cls._deriv_fetcher @classmethod def fetch_data(cls, symbol_key: str, period: str = "6mo", interval: str = "1h") -> pd.DataFrame: deriv_pairs = DerivDataFetcher.get_available_pairs() if symbol_key in deriv_pairs: return cls._get_deriv_fetcher().fetch_data(symbol_key, period, interval) ticker = cls.YAHOO_PAIRS.get(symbol_key, symbol_key) try: df = yf.download(ticker, period=period, interval=interval, progress=False) if df.empty: return pd.DataFrame() if isinstance(df.columns, pd.MultiIndex): df.columns = [col[0].lower() for col in df.columns] else: df.columns = [col.lower() for col in df.columns] return df except Exception as e: print(f"❌ Error fetching {symbol_key}: {e}") return pd.DataFrame() @classmethod def get_available_pairs(cls) -> List[str]: yahoo_pairs = list(cls.YAHOO_PAIRS.keys()) deriv_pairs = DerivDataFetcher.get_available_pairs() return yahoo_pairs + deriv_pairs