feat(data): 实现数据获取层抽象接口

- OHLCVData: 标准化K线数据结构
- DataSource: 数据源抽象接口(fetch/fetch_batch)
- DataCache: 缓存抽象接口(get/set/is_fresh)
- LocalFileCache: 本地文件缓存实现
- HybridDataSourceAdapter/TushareDataSource/YFinanceDataSource: 定制数据源适配器
This commit is contained in:
2026-05-11 23:24:11 +08:00
parent c5a41b71ae
commit 774758c3b0
5 changed files with 375 additions and 0 deletions

126
framework/data/__init__.py Normal file
View File

@@ -0,0 +1,126 @@
"""
数据层抽象接口(通用)
只提供数据获取抽象接口具体实现在strategies/shared/data/
"""
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any
import pandas as pd
from dataclasses import dataclass
from datetime import datetime
@dataclass
class OHLCVData:
"""
OHLCV数据结构通用
标准化的K线数据格式
"""
code: str
name: str = ""
start_date: datetime = None
end_date: datetime = None
# OHLCV数据DataFrame
data: pd.DataFrame = None
@property
def length(self) -> int:
"""数据长度"""
return len(self.data) if self.data is not None else 0
def validate(self) -> bool:
"""验证数据完整性"""
if self.data is None or self.data.empty:
return False
required_cols = ['close']
return all(col in self.data.columns for col in required_cols)
def __repr__(self) -> str:
return f"OHLCVData(code={self.code}, name={self.name}, length={self.length})"
class DataSource(ABC):
"""
数据源抽象接口
所有数据源必须实现fetch方法
"""
name: str = "base"
def __init__(self, **params):
"""初始化数据源参数"""
self._params = params
@abstractmethod
def fetch(self, code: str, start: str, end: str) -> OHLCVData:
"""
获取单个标的的OHLCV数据
Args:
code: 标的代码
start: 开始日期 (YYYY-MM-DD)
end: 结束日期 (YYYY-MM-DD)
Returns:
OHLCVData对象
"""
pass
@abstractmethod
def fetch_batch(self, codes: List[str], start: str, end: str) -> Dict[str, OHLCVData]:
"""
批量获取多个标的的OHLCV数据
Args:
codes: 标的代码列表
start: 开始日期
end: 结束日期
Returns:
{code: OHLCVData}字典
"""
pass
def get_supported_codes(self) -> List[str]:
"""获取支持的数据源代码列表"""
return []
def __repr__(self) -> str:
return f"{self.__class__.__name__}(name={self.name})"
class DataCache(ABC):
"""
数据缓存抽象接口(通用)
支持本地缓存管理
"""
@abstractmethod
def get(self, code: str, start: str, end: str) -> Optional[OHLCVData]:
"""从缓存获取数据"""
pass
@abstractmethod
def set(self, code: str, data: OHLCVData) -> None:
"""写入缓存"""
pass
@abstractmethod
def is_fresh(self, code: str, max_age_days: int = 1) -> bool:
"""检查缓存是否新鲜"""
pass
@abstractmethod
def clear(self, code: Optional[str] = None) -> None:
"""清空缓存"""
pass
# 导出抽象接口
__all__ = ['OHLCVData', 'DataSource', 'DataCache']