aszerW
908b28473f
## 架构设计 - 三层架构：core(抽象接口) → shared(通用实现) → tests(验证测试) - 5个核心抽象基类：StrategyBase, FactorBase, SignalGenerator, Executor, DataFetcher - 零侵入：与现有框架并行开发，不修改生产代码 ## 已完成 ✓ 核心接口层（5个ABC类） ✓ 通用因子层（MomentumFactor完全复制现有逻辑） ✓ 对比验证测试（新旧因子输出差异=0，测试通过） ## 验证结果 - 最大差异: 0.000000e+00 - 平均差异: 0.000000e+00 - 容差: < 1e-10 ## 下一步 - 阶段3: 信号层迁移（TopNSelector, DynamicThreshold, RebalanceController） - 阶段4: 执行层迁移（BacktestRunner） - 阶段5: 数据层迁移（DataFetcher实现） - 阶段6: 完整策略对比验证 ## 设计原则 - 按需抽象，不预先设计 - 职责分离，避免框架膨胀 - 测试驱动，每个组件必须有对比测试 - 渐进式迁移，验证通过再替换
98 lines
2.2 KiB
Python
98 lines
2.2 KiB
Python
"""
|
|
数据获取器抽象基类
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Dict, List, Optional
|
|
import pandas as pd
|
|
|
|
|
|
class DataFetcher(ABC):
|
|
"""
|
|
数据获取器抽象基类
|
|
|
|
所有数据获取器必须实现必要方法
|
|
"""
|
|
|
|
name: str = "base"
|
|
|
|
def __init__(self, **params):
|
|
"""
|
|
初始化数据获取器参数
|
|
|
|
Args:
|
|
**params: 数据源参数（如 api_url, ssh_config 等）
|
|
"""
|
|
self._params = params
|
|
|
|
@abstractmethod
|
|
def fetch_indices(
|
|
self,
|
|
codes: List[str],
|
|
start: str,
|
|
end: str
|
|
) -> Dict[str, pd.DataFrame]:
|
|
"""
|
|
获取指数 OHLCV 数据
|
|
|
|
Args:
|
|
codes: 指数代码列表
|
|
start: 开始日期 (YYYY-MM-DD)
|
|
end: 结束日期 (YYYY-MM-DD)
|
|
|
|
Returns:
|
|
{code: DataFrame} 字典
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def fetch_etf(
|
|
self,
|
|
codes: List[str],
|
|
start: str,
|
|
end: str
|
|
) -> Dict[str, pd.DataFrame]:
|
|
"""
|
|
获取 ETF 数据（价格 + 净值）
|
|
|
|
Args:
|
|
codes: ETF 代码列表
|
|
start: 开始日期
|
|
end: 结束日期
|
|
|
|
Returns:
|
|
{code: DataFrame} 字典
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_trading_calendar(self, market: str = 'A') -> pd.Index:
|
|
"""
|
|
获取交易日历
|
|
|
|
Args:
|
|
market: 市场代码（'A', 'US', 'HK' 等）
|
|
|
|
Returns:
|
|
交易日历 Index
|
|
"""
|
|
pass
|
|
|
|
def get_benchmark(self, code: str, start: str, end: str) -> pd.Series:
|
|
"""
|
|
获取基准数据（可选）
|
|
|
|
Args:
|
|
code: 基准代码
|
|
start: 开始日期
|
|
end: 结束日期
|
|
|
|
Returns:
|
|
基准收盘价 Series
|
|
"""
|
|
raise NotImplementedError("Optional method")
|
|
|
|
def __repr__(self) -> str:
|
|
params_str = ', '.join([f"{k}={v}" for k, v in self._params.items()])
|
|
return f"{self.__class__.__name__}(name={self.name})"
|