futures-data-platform/backend/app/services/datasource/base.py

from abc import ABC, abstractmethod
from typing import List, Optional
from datetime import datetime
import pandas as pd


class DataSourceBase(ABC):
    """数据源基类，所有数据源适配器必须实现这些接口"""

    def __init__(self, config: dict):
        self.config = config
        self.name = self.__class__.__name__
        self._initialized = False

    @abstractmethod
    def initialize(self) -> bool:
        """初始化数据源连接，返回是否成功"""
        pass

    @abstractmethod
    def get_contract_list(self, exchange: Optional[str] = None) -> List[dict]:
        """
        获取合约列表
        返回: [{"symbol": "rb2401", "exchange": "SHFE", "name": "螺纹钢2401", ...}]
        """
        pass

    @abstractmethod
    def get_kline_daily(
        self,
        symbol: str,
        start_date: str,
        end_date: str
    ) -> pd.DataFrame:
        """
        获取日K线数据
        返回 DataFrame 包含: trade_date, open, high, low, close, volume, turnover, open_interest, settle, pre_settle
        """
        pass

    @abstractmethod
    def get_kline_weekly(
        self,
        symbol: str,
        start_date: str,
        end_date: str
    ) -> pd.DataFrame:
        """
        获取周K线数据
        返回 DataFrame 包含: trade_date, open, high, low, close, volume, turnover, open_interest
        """
        pass

    @abstractmethod
    def get_kline_intraday(
        self,
        symbol: str,
        period: str,  # 5m/15m/30m/60m
        start_date: str,
        end_date: str
    ) -> pd.DataFrame:
        """
        获取分钟级K线数据
        返回 DataFrame 包含: trade_time, open, high, low, close, volume, turnover, open_interest
        """
        pass

    def health_check(self) -> tuple[bool, str]:
        """健康检查，返回 (是否健康, 错误信息)"""
        try:
            ok = self.initialize()
            if ok:
                return True, ""
            return False, "初始化失败"
        except Exception as e:
            return False, str(e)