ff7dc0c95a
feat(frontend): integrate Prisma and reports API/pages chore(config): add data_sources.yaml; update analysis-config.json docs: add 2025-11-03 dev log; update user guide scripts: enhance dev.sh; add tushare_legacy_client deps: update backend and frontend dependencies
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
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]) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetches financial statements for a list of report dates.
|
|
|
|
This method should aim to fetch data for all requested dates in a single call if possible
|
|
and then combine them into a unified format.
|
|
|
|
:param stock_code: The stock identifier.
|
|
:param report_dates: A list of report dates to fetch data for (e.g., ['20231231', '20221231']).
|
|
:return: A list of dictionaries, each containing financial statement data for a specific period.
|
|
"""
|
|
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.
|
|
|
|
: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.
|
|
"""
|
|
results = await self.get_financial_statements(stock_code, [report_date])
|
|
return results[0] if results else None
|