Pascal
c462342d44
- Add base API client and cache manager classes - Implement FMP and YFinance specific clients and cache managers - Add API factory for managing multiple data providers - Add test suite for API configuration and caching - Add logging configuration for API operations
108 lines
3.0 KiB
Python
108 lines
3.0 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Dict, List, Optional, Union
|
|
import pandas as pd
|
|
from datetime import datetime
|
|
import time
|
|
|
|
class BaseAPIClient(ABC):
|
|
"""Base class for all API clients."""
|
|
|
|
def __init__(self, api_key: Optional[str] = None):
|
|
self.api_key = api_key
|
|
self.last_request_time = None
|
|
self.rate_limit_delay = 1.0 # Default 1 second between requests
|
|
|
|
@abstractmethod
|
|
def get_etf_profile(self, symbol: str) -> Dict:
|
|
"""Get ETF profile data.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
|
|
Returns:
|
|
Dictionary containing ETF profile information
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
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
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
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
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_dividend_history(self, symbol: str) -> pd.DataFrame:
|
|
"""Get dividend history.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
|
|
Returns:
|
|
DataFrame with dividend history
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_sector_weightings(self, symbol: str) -> Dict:
|
|
"""Get sector weightings.
|
|
|
|
Args:
|
|
symbol: ETF ticker symbol
|
|
|
|
Returns:
|
|
Dictionary with sector weightings
|
|
"""
|
|
pass
|
|
|
|
def _check_rate_limit(self):
|
|
"""Check and enforce rate limiting."""
|
|
if self.last_request_time:
|
|
time_since_last = (datetime.now() - self.last_request_time).total_seconds()
|
|
if time_since_last < self.rate_limit_delay:
|
|
time.sleep(self.rate_limit_delay - time_since_last)
|
|
self.last_request_time = datetime.now()
|
|
|
|
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 _handle_error(self, error: Exception) -> Dict:
|
|
"""Handle API errors consistently.
|
|
|
|
Args:
|
|
error: Exception that occurred
|
|
|
|
Returns:
|
|
Dictionary with error information
|
|
"""
|
|
return {
|
|
'error': True,
|
|
'message': str(error),
|
|
'timestamp': datetime.now().isoformat()
|
|
} |