149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
import yfinance as yf
|
|
import pandas as pd
|
|
from typing import Dict, List, Optional
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
from ..base import BaseAPIClient
|
|
from ...cache.yfinance_cache import YFinanceCacheManager
|
|
|
|
class YFinanceClient(BaseAPIClient):
|
|
"""Yahoo Finance API client."""
|
|
|
|
def __init__(self, cache_manager: Optional[YFinanceCacheManager] = None):
|
|
"""Initialize YFinance client.
|
|
|
|
Args:
|
|
cache_manager: Optional cache manager instance
|
|
"""
|
|
super().__init__()
|
|
self.cache_manager = cache_manager or YFinanceCacheManager()
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
|
|
def _validate_symbol(self, symbol: str) -> bool:
|
|
"""Validate ETF symbol format.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
|
|
Returns:
|
|
True if valid, False otherwise
|
|
"""
|
|
return bool(symbol and isinstance(symbol, str) and symbol.isupper())
|
|
|
|
def get_etf_info(self, symbol: str) -> Dict:
|
|
"""Get ETF information.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
|
|
Returns:
|
|
Dictionary with ETF information
|
|
"""
|
|
if not self._validate_symbol(symbol):
|
|
return self._handle_error(ValueError(f"Invalid symbol: {symbol}"))
|
|
|
|
# Try cache first
|
|
if self.cache_manager:
|
|
is_valid, cached_data = self.cache_manager.load('yfinance', symbol, 'info')
|
|
if is_valid:
|
|
return cached_data
|
|
|
|
try:
|
|
etf = yf.Ticker(symbol)
|
|
info = etf.info
|
|
|
|
# Cache the response
|
|
if self.cache_manager and info:
|
|
self.cache_manager.save('yfinance', symbol, 'info', info)
|
|
|
|
return info
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching ETF info: {str(e)}")
|
|
return self._handle_error(e)
|
|
|
|
def get_historical_data(self, symbol: str, period: str = '1y', interval: str = '1d') -> pd.DataFrame:
|
|
"""Get historical price data.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
period: Time period (e.g., '1d', '1w', '1m', '1y')
|
|
interval: Data interval (e.g., '1d', '1wk', '1mo')
|
|
|
|
Returns:
|
|
DataFrame with historical price data
|
|
"""
|
|
if not self._validate_symbol(symbol):
|
|
return pd.DataFrame()
|
|
|
|
# Try cache first
|
|
if self.cache_manager:
|
|
is_valid, cached_data = self.cache_manager.load('yfinance', symbol, f'historical_{period}_{interval}')
|
|
if is_valid:
|
|
return pd.DataFrame(cached_data)
|
|
|
|
try:
|
|
etf = yf.Ticker(symbol)
|
|
data = etf.history(period=period, interval=interval)
|
|
|
|
# Cache the response
|
|
if self.cache_manager and not data.empty:
|
|
self.cache_manager.save('yfinance', symbol, f'historical_{period}_{interval}', data.to_dict('records'))
|
|
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching historical data: {str(e)}")
|
|
return pd.DataFrame()
|
|
|
|
def get_dividend_history(self, symbol: str) -> pd.DataFrame:
|
|
"""Get dividend history.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
|
|
Returns:
|
|
DataFrame with dividend history
|
|
"""
|
|
if not self._validate_symbol(symbol):
|
|
return pd.DataFrame()
|
|
|
|
# Try cache first
|
|
if self.cache_manager:
|
|
is_valid, cached_data = self.cache_manager.load('yfinance', symbol, 'dividend_history')
|
|
if is_valid:
|
|
return pd.DataFrame(cached_data)
|
|
|
|
try:
|
|
etf = yf.Ticker(symbol)
|
|
data = etf.dividends.to_frame()
|
|
|
|
# Cache the response
|
|
if self.cache_manager and not data.empty:
|
|
self.cache_manager.save('yfinance', symbol, 'dividend_history', data.to_dict('records'))
|
|
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching dividend history: {str(e)}")
|
|
return pd.DataFrame()
|
|
|
|
def clear_cache(self) -> int:
|
|
"""Clear expired cache entries.
|
|
|
|
Returns:
|
|
Number of files cleared
|
|
"""
|
|
if self.cache_manager:
|
|
return self.cache_manager.clear_expired()
|
|
return 0
|
|
|
|
def get_cache_stats(self) -> Dict:
|
|
"""Get cache statistics.
|
|
|
|
Returns:
|
|
Dictionary with cache statistics
|
|
"""
|
|
if self.cache_manager:
|
|
return self.cache_manager.get_stats()
|
|
return {} |