ETF_Suite_Portal/ETF_Portal/api/fmp/client.py

import requests
import pandas as pd
from typing import Dict, List, Optional
from datetime import datetime
import logging
from ..base import BaseAPIClient
from ...cache.fmp_cache import FMPCacheManager

class FMPClient(BaseAPIClient):
    """Financial Modeling Prep API client."""
    
    BASE_URL = "https://financialmodelingprep.com/api/v3"
    
    def __init__(self, api_key: str, cache_manager: Optional[FMPCacheManager] = None):
        """Initialize FMP client.
        
        Args:
            api_key: FMP API key
            cache_manager: Optional cache manager instance
        """
        super().__init__(api_key)
        self.cache_manager = cache_manager or FMPCacheManager()
        self.logger = logging.getLogger(self.__class__.__name__)
    
    def _make_request(self, endpoint: str, params: Dict = None) -> Dict:
        """Make API request to FMP.
        
        Args:
            endpoint: API endpoint
            params: Query parameters
            
        Returns:
            API response data
        """
        # Check cache first
        if self.cache_manager:
            cached_data, is_valid = self.cache_manager.get(endpoint, params)
            if is_valid:
                return cached_data
        
        # Prepare request
        url = f"{self.BASE_URL}/{endpoint}"
        params = params or {}
        params['apikey'] = self.api_key
        
        # Check rate limit
        self._check_rate_limit()
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            data = response.json()
            
            # Cache the response
            if self.cache_manager:
                self.cache_manager.set(endpoint, data, params)
            
            return data
            
        except requests.exceptions.RequestException as e:
            self.logger.error(f"FMP API request failed: {str(e)}")
            return self._handle_error(e)
    
    def get_etf_profile(self, symbol: str) -> Dict:
        """Get ETF profile data.
        
        Args:
            symbol: ETF ticker symbol
            
        Returns:
            Dictionary containing ETF profile information
        """
        if not self._validate_symbol(symbol):
            return self._handle_error(ValueError(f"Invalid symbol: {symbol}"))
        
        return self._make_request(f"etf/profile/{symbol}")
    
    def get_etf_holdings(self, symbol: str) -> List[Dict]:
        """Get ETF holdings data.
        
        Args:
            symbol: ETF ticker symbol
            
        Returns:
            List of dictionaries containing holding information
        """
        if not self._validate_symbol(symbol):
            return [self._handle_error(ValueError(f"Invalid symbol: {symbol}"))]
        
        return self._make_request(f"etf/holdings/{symbol}")
    
    def get_historical_data(self, symbol: str, period: str = '1y') -> pd.DataFrame:
        """Get historical price data.
        
        Args:
            symbol: ETF ticker symbol
            period: Time period (e.g., '1d', '1w', '1m', '1y')
            
        Returns:
            DataFrame with historical price data
        """
        if not self._validate_symbol(symbol):
            return pd.DataFrame()
        
        data = self._make_request(f"etf/historical-price/{symbol}", {'period': period})
        
        if isinstance(data, dict) and data.get('error'):
            return pd.DataFrame()
        
        return pd.DataFrame(data)
    
    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()
        
        data = self._make_request(f"etf/dividend/{symbol}")
        
        if isinstance(data, dict) and data.get('error'):
            return pd.DataFrame()
        
        return pd.DataFrame(data)
    
    def get_sector_weightings(self, symbol: str) -> Dict:
        """Get sector weightings.
        
        Args:
            symbol: ETF ticker symbol
            
        Returns:
            Dictionary with sector weightings
        """
        if not self._validate_symbol(symbol):
            return self._handle_error(ValueError(f"Invalid symbol: {symbol}"))
        
        return self._make_request(f"etf/sector-weightings/{symbol}")
    
    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 {}