from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseDataProvider(ABC): """ Abstract base class for all financial data providers. """ def __init__(self, token: Optional[str] = None): """ Initializes the data provider, optionally with an API token. :param token: API token for the data provider, if required. """ self.token = token self._initialize() def _initialize(self): """ Perform any necessary initialization, such as API client setup. This method is called by the constructor. """ pass @abstractmethod async def get_stock_basic(self, stock_code: str) -> Optional[Dict[str, Any]]: """ Fetches basic company information for a given stock code. :param stock_code: The stock identifier. :return: A dictionary with basic company info, or None if not found. """ pass @abstractmethod async def get_daily_price(self, stock_code: str, start_date: str, end_date: str) -> List[Dict[str, Any]]: """ Fetches daily stock prices for a given period. :param stock_code: The stock identifier. :param start_date: The start date of the period (e.g., 'YYYYMMDD'). :param end_date: The end date of the period (e.g., 'YYYYMMDD'). :return: A list of dictionaries, each representing a day's price data. """ pass @abstractmethod async def get_financial_statements(self, stock_code: str, report_dates: List[str]) -> Dict[str, List[Dict[str, Any]]]: """ Fetches financial statements for a list of report dates and returns them in a series format. The series format is a dictionary where keys are metric names (e.g., 'revenue') and values are a list of data points over time. e.g., {"revenue": [{"year": "2023", "value": 1000}, ...]} Providers should also calculate derived metrics if they are not directly available. :param stock_code: The stock identifier. :param report_dates: A list of report dates to fetch data for (e.g., ['20231231', '20221231']). :return: A dictionary in series format. """ pass async def get_financial_statement(self, stock_code: str, report_date: str) -> Optional[Dict[str, Any]]: """ Fetches a single financial statement for a specific report date. This is a convenience method that can be implemented by calling get_financial_statements. Note: The return value of this function is a single report (dictionary), not a series object. This is for compatibility with parts of the code that need a single flat report. :param stock_code: The stock identifier. :param report_date: The report date for the statement (e.g., '20231231'). :return: A dictionary with financial statement data, or None if not found. """ series_data = await self.get_financial_statements(stock_code, [report_date]) if not series_data: return None report: Dict[str, Any] = {"ts_code": stock_code, "end_date": report_date} for metric, points in series_data.items(): for point in points: if point.get("year") == report_date[:4]: report[metric] = point.get("value") break return report