"""Base abstract interface for market data providers. This module defines the abstract interface that all market data providers must implement, ensuring consistent data access patterns across different providers. """ from abc import ABC, abstractmethod from typing import Dict, List, Optional, Any from datetime import datetime from decimal import Decimal import pandas as pd from pydantic import BaseModel, Field, ConfigDict, field_serializer class QuoteData(BaseModel): """Real-time or latest quote data for a security. Attributes: symbol: Stock ticker symbol price: Current or latest price change: Absolute price change change_percent: Percentage price change volume: Trading volume market_cap: Market capitalisation timestamp: Quote timestamp open: Opening price high: High price low: Low price previous_close: Previous closing price """ symbol: str price: Decimal change: Optional[Decimal] = None change_percent: Optional[Decimal] = None volume: Optional[int] = None market_cap: Optional[Decimal] = None timestamp: datetime open: Optional[Decimal] = None high: Optional[Decimal] = None low: Optional[Decimal] = None previous_close: Optional[Decimal] = None model_config = ConfigDict() @field_serializer('price', 'change', 'change_percent', 'market_cap', 'open', 'high', 'low', 'previous_close', mode='plain') def serialize_decimal(self, value: Optional[Decimal]) -> Optional[str]: """Serialise Decimal values as strings.""" if value is None: return None return str(value) @field_serializer('timestamp', mode='plain') def serialize_datetime(self, value: datetime) -> str: """Serialise datetime values as ISO format strings.""" return value.isoformat() class CompanyProfile(BaseModel): """Company profile and fundamental information. Attributes: symbol: Stock ticker symbol company_name: Company name description: Company description sector: Industry sector industry: Specific industry country: Country of incorporation exchange: Primary stock exchange currency: Trading currency market_cap: Market capitalisation employees: Number of employees website: Company website ceo: CEO name founded: Year founded """ symbol: str company_name: str description: Optional[str] = None sector: Optional[str] = None industry: Optional[str] = None country: Optional[str] = None exchange: Optional[str] = None currency: Optional[str] = None market_cap: Optional[Decimal] = None employees: Optional[int] = None website: Optional[str] = None ceo: Optional[str] = None founded: Optional[int] = None class FinancialRatios(BaseModel): """Financial ratios and metrics. Attributes: symbol: Stock ticker symbol pe_ratio: Price to Earnings ratio pb_ratio: Price to Book ratio ps_ratio: Price to Sales ratio dividend_yield: Dividend yield percentage payout_ratio: Dividend payout ratio roe: Return on Equity roa: Return on Assets debt_to_equity: Debt to Equity ratio current_ratio: Current ratio quick_ratio: Quick ratio """ symbol: str pe_ratio: Optional[Decimal] = None pb_ratio: Optional[Decimal] = None ps_ratio: Optional[Decimal] = None dividend_yield: Optional[Decimal] = None payout_ratio: Optional[Decimal] = None roe: Optional[Decimal] = None roa: Optional[Decimal] = None debt_to_equity: Optional[Decimal] = None current_ratio: Optional[Decimal] = None quick_ratio: Optional[Decimal] = None class MarketDataProvider(ABC): """Abstract base class for market data providers. All market data providers must implement this interface to ensure consistent data access patterns and enable easy provider switching. Usage: class MyProvider(MarketDataProvider): def get_historical_prices(self, symbol, start_date, end_date): # Implementation pass """ @abstractmethod def get_quote(self, symbol: str) -> QuoteData: """Fetch real-time or latest quote for a symbol. Args: symbol: Stock ticker symbol (e.g., "AAPL") Returns: QuoteData: Real-time quote information Raises: ValueError: If symbol is invalid RuntimeError: If API request fails """ pass @abstractmethod def get_quotes(self, symbols: List[str]) -> Dict[str, QuoteData]: """Fetch quotes for multiple symbols. Args: symbols: List of stock ticker symbols Returns: Dict mapping symbols to QuoteData Raises: ValueError: If any symbol is invalid RuntimeError: If API request fails """ pass @abstractmethod def get_historical_prices( self, symbol: str, start_date: str, end_date: str, interval: str = "1d" ) -> pd.DataFrame: """Fetch historical OHLCV data. Args: symbol: Stock ticker symbol start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format interval: Data interval (1d, 1h, 1m, etc.) Returns: DataFrame with columns: date, open, high, low, close, volume Index is date, all prices are Decimal type Raises: ValueError: If dates are invalid or symbol not found RuntimeError: If API request fails """ pass @abstractmethod def get_company_profile(self, symbol: str) -> CompanyProfile: """Fetch company profile information. Args: symbol: Stock ticker symbol Returns: CompanyProfile: Company information Raises: ValueError: If symbol is invalid RuntimeError: If API request fails """ pass @abstractmethod def get_financial_ratios(self, symbol: str) -> FinancialRatios: """Fetch financial ratios and metrics. Args: symbol: Stock ticker symbol Returns: FinancialRatios: Financial metrics Raises: ValueError: If symbol is invalid RuntimeError: If API request fails """ pass @abstractmethod def get_technical_indicators( self, symbol: str, start_date: str, end_date: str, indicators: List[str] ) -> Dict[str, pd.DataFrame]: """Calculate technical indicators. Args: symbol: Stock ticker symbol start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format indicators: List of indicator names (e.g., ["RSI", "MACD", "BB"]) Returns: Dict mapping indicator names to DataFrames with calculated values Raises: ValueError: If dates or indicators are invalid RuntimeError: If API request fails """ pass @property @abstractmethod def name(self) -> str: """Provider name for logging and identification. Returns: str: Provider name (e.g., "YFinance", "FMP", "EODHD") """ pass @property @abstractmethod def rate_limit_info(self) -> Dict[str, Any]: """Get current rate limit information. Returns: Dict with keys: - requests_per_minute: Maximum requests per minute - requests_remaining: Requests remaining in current window - window_reset: Timestamp when window resets """ pass